from __future__ import annotations import datetime as dt from typing import Any, Dict, List, Optional import httpx from pydantic import BaseModel from app.core.config import settings class CompanyProfile(BaseModel): symbol: str name: str industry: Optional[str] = None list_date: Optional[dt.date] = None additional_info: Optional[Dict[str, Any]] = None class TimeSeriesFinancial(BaseModel): symbol: str metric_name: str period_date: dt.date value: float source: Optional[str] = None class TimeSeriesFinancialBatch(BaseModel): records: List[TimeSeriesFinancial] class DailyMarketData(BaseModel): symbol: str trade_date: dt.date open_price: Optional[float] = None high_price: Optional[float] = None low_price: Optional[float] = None close_price: Optional[float] = None volume: Optional[int] = None pe: Optional[float] = None pb: Optional[float] = None total_mv: Optional[float] = None class DailyMarketDataBatch(BaseModel): records: List[DailyMarketData] class RealtimeQuote(BaseModel): symbol: str market: str ts: dt.datetime price: float open_price: Optional[float] = None high_price: Optional[float] = None low_price: Optional[float] = None prev_close: Optional[float] = None change: Optional[float] = None change_percent: Optional[float] = None volume: Optional[int] = None source: Optional[str] = None class NewAnalysisResult(BaseModel): symbol: str module_id: str model_name: Optional[str] = None content: str meta_data: Optional[Dict[str, Any]] = None class AnalysisResult(BaseModel): id: str symbol: str module_id: str generated_at: dt.datetime model_name: Optional[str] = None content: str meta_data: Optional[Dict[str, Any]] = None class DataPersistenceClient: def __init__(self, base_url: Optional[str] = None, timeout: float = 20.0): self.base_url = (base_url or settings.DATA_PERSISTENCE_BASE_URL).rstrip("/") self.timeout = timeout async def _client(self) -> httpx.AsyncClient: return httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout) # Companies async def upsert_company(self, profile: CompanyProfile) -> None: async with await self._client() as client: resp = await client.put("/companies", json=profile.model_dump(mode="json")) resp.raise_for_status() async def get_company(self, symbol: str) -> CompanyProfile: async with await self._client() as client: resp = await client.get(f"/companies/{symbol}") resp.raise_for_status() return CompanyProfile.model_validate(resp.json()) # Financials async def batch_insert_financials(self, batch: TimeSeriesFinancialBatch) -> None: async with await self._client() as client: resp = await client.post("/market-data/financials/batch", json=batch.model_dump(mode="json")) resp.raise_for_status() async def get_financials_by_symbol(self, symbol: str, metrics: Optional[List[str]] = None) -> List[TimeSeriesFinancial]: params = {} if metrics: params["metrics"] = ",".join(metrics) async with await self._client() as client: resp = await client.get(f"/market-data/financials/{symbol}", params=params) resp.raise_for_status() return [TimeSeriesFinancial.model_validate(item) for item in resp.json()] # Daily data async def batch_insert_daily_data(self, batch: DailyMarketDataBatch) -> None: async with await self._client() as client: resp = await client.post("/market-data/daily/batch", json=batch.model_dump(mode="json")) resp.raise_for_status() async def get_daily_data_by_symbol( self, symbol: str, start_date: Optional[dt.date] = None, end_date: Optional[dt.date] = None, ) -> List[DailyMarketData]: params = {} if start_date: params["start_date"] = start_date.isoformat() if end_date: params["end_date"] = end_date.isoformat() async with await self._client() as client: resp = await client.get(f"/market-data/daily/{symbol}", params=params) resp.raise_for_status() return [DailyMarketData.model_validate(item) for item in resp.json()] # Realtime quotes async def upsert_realtime_quote(self, quote: RealtimeQuote) -> None: async with await self._client() as client: resp = await client.post("/market-data/quotes", json=quote.model_dump(mode="json")) resp.raise_for_status() async def get_latest_realtime_quote( self, market: str, symbol: str, max_age_seconds: Optional[int] = None, ) -> Optional[RealtimeQuote]: params = {"market": market} if max_age_seconds is not None: params["max_age_seconds"] = int(max_age_seconds) async with await self._client() as client: resp = await client.get(f"/market-data/quotes/{symbol}", params=params) if resp.status_code == 404: return None resp.raise_for_status() return RealtimeQuote.model_validate(resp.json()) # Analysis results async def create_analysis_result(self, new_result: NewAnalysisResult) -> AnalysisResult: async with await self._client() as client: resp = await client.post("/analysis-results", json=new_result.model_dump(mode="json")) resp.raise_for_status() return AnalysisResult.model_validate(resp.json()) async def get_analysis_results(self, symbol: str, module_id: Optional[str] = None) -> List[AnalysisResult]: params = {"symbol": symbol} if module_id: params["module_id"] = module_id async with await self._client() as client: resp = await client.get("/analysis-results", params=params) resp.raise_for_status() return [AnalysisResult.model_validate(item) for item in resp.json()] async def get_analysis_result_by_id(self, result_id: str) -> AnalysisResult: async with await self._client() as client: resp = await client.get(f"/analysis-results/{result_id}") resp.raise_for_status() return AnalysisResult.model_validate(resp.json())