""" Database statistics and status endpoints """ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func, distinct, text from typing import Dict, Any, List import logging from datetime import datetime from app.core.database import get_db from app.models.financial import Company, FinancialData, CalculatedMetrics, PriceData from app.models.etf import ETFHoldingsSnapshot, ETFHolding from app.schemas.financial import DataSource router = APIRouter() logger = logging.getLogger(__name__) @router.get("/stats") async def get_database_stats(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]: """ 데이터베이스 통계 정보를 반환합니다. Returns: Dict containing database statistics including: - companies: total, with_financial_data, with_price_data - financial_data: total_records, real_data, estimated_data, date_range, by_source - price_data: total_records, date_range, tickers - calculated_metrics: total_records, date_range """ try: logger.info("Fetching database statistics") # 회사 통계 companies_total = await db.execute(select(func.count(Company.id))) companies_total = companies_total.scalar() # 재무 데이터가 있는 회사 수 companies_with_financial = await db.execute( select(func.count(distinct(FinancialData.ticker))) ) companies_with_financial = companies_with_financial.scalar() # 주가 데이터가 있는 회사 수 companies_with_price = await db.execute( select(func.count(distinct(PriceData.ticker))) ) companies_with_price = companies_with_price.scalar() # 재무 데이터 통계 financial_total = await db.execute(select(func.count(FinancialData.id))) financial_total = financial_total.scalar() financial_real = await db.execute( select(func.count(FinancialData.id)).where(FinancialData.is_estimated == False) ) financial_real = financial_real.scalar() financial_estimated = await db.execute( select(func.count(FinancialData.id)).where(FinancialData.is_estimated == True) ) financial_estimated = financial_estimated.scalar() # 재무 데이터 날짜 범위 financial_date_range = await db.execute( select( func.min(FinancialData.period_date), func.max(FinancialData.period_date) ) ) financial_dates = financial_date_range.first() # 데이터 소스별 분포 financial_by_source = await db.execute( select( FinancialData.data_source, func.count(FinancialData.id) ).group_by(FinancialData.data_source) ) source_distribution = {source: count for source, count in financial_by_source.all()} # 주가 데이터 통계 price_total = await db.execute(select(func.count(PriceData.id))) price_total = price_total.scalar() # 주가 데이터 날짜 범위 price_date_range = await db.execute( select( func.min(PriceData.date), func.max(PriceData.date) ) ) price_dates = price_date_range.first() # 주가 데이터 종목 목록 price_tickers = await db.execute( select(distinct(PriceData.ticker)).order_by(PriceData.ticker) ) ticker_list = [ticker for ticker, in price_tickers.all()] # 계산된 지표 통계 metrics_total = await db.execute(select(func.count(CalculatedMetrics.id))) metrics_total = metrics_total.scalar() # 계산된 지표 날짜 범위 metrics_date_range = await db.execute( select( func.min(CalculatedMetrics.period_date), func.max(CalculatedMetrics.period_date) ) ) metrics_dates = metrics_date_range.first() # 결과 구성 stats = { "companies": { "total": companies_total or 0, "with_financial_data": companies_with_financial or 0, "with_price_data": companies_with_price or 0 }, "financial_data": { "total_records": financial_total or 0, "real_data": financial_real or 0, "estimated_data": financial_estimated or 0, "date_range": { "earliest": financial_dates[0].isoformat() if financial_dates[0] else None, "latest": financial_dates[1].isoformat() if financial_dates[1] else None }, "by_source": source_distribution }, "price_data": { "total_records": price_total or 0, "date_range": { "earliest": price_dates[0].isoformat() if price_dates[0] else None, "latest": price_dates[1].isoformat() if price_dates[1] else None }, "tickers": ticker_list }, "calculated_metrics": { "total_records": metrics_total or 0, "date_range": { "earliest": metrics_dates[0].isoformat() if metrics_dates[0] else None, "latest": metrics_dates[1].isoformat() if metrics_dates[1] else None } } } logger.info(f"Successfully fetched database statistics: {financial_total} financial records, {price_total} price records") return stats except Exception as e: logger.error(f"Error fetching database statistics: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to fetch database statistics: {str(e)}" ) @router.get("/health") async def get_database_health(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]: """ 데이터베이스 연결 상태를 확인합니다. """ try: # 간단한 쿼리를 실행하여 연결 상태 확인 result = await db.execute(select(1)) result.scalar() return { "status": "healthy", "database": "connected", "timestamp": "2025-08-02T12:00:00Z" } except Exception as e: logger.error(f"Database health check failed: {str(e)}") raise HTTPException( status_code=503, detail=f"Database connection failed: {str(e)}" ) @router.get("/tables") async def get_table_info(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]: """ 데이터베이스 테이블 정보를 반환합니다. """ try: table_info = {} # 각 테이블의 레코드 수 조회 tables = [ ("companies", Company), ("financial_data", FinancialData), ("price_data", PriceData), ("calculated_metrics", CalculatedMetrics) ] for table_name, model in tables: count = await db.execute(select(func.count(model.id))) table_info[table_name] = { "record_count": count.scalar() or 0, "table_name": table_name } return { "tables": table_info, "total_tables": len(table_info) } except Exception as e: logger.error(f"Error fetching table info: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to fetch table info: {str(e)}" ) @router.post("/cleanup/duplicates") async def cleanup_duplicate_records(db: AsyncSession = Depends(get_db)) -> Dict[str, Any]: """ Remove duplicate financial and metrics records, keeping the most recent real data. """ try: logger.info("Starting duplicate record cleanup") # Start transaction async with db.begin(): # Clean financial data duplicates - keep most recent real data financial_cleanup = await db.execute(text(""" DELETE FROM financial_data WHERE id NOT IN ( SELECT DISTINCT ON (ticker, period_date, period_type) id FROM financial_data ORDER BY ticker, period_date, period_type, CASE WHEN is_estimated = false THEN 0 ELSE 1 END, -- Real data first created_at DESC -- Most recent first ) """)) # Clean calculated metrics duplicates metrics_cleanup = await db.execute(text(""" DELETE FROM calculated_metrics WHERE id NOT IN ( SELECT DISTINCT ON (ticker, period_date) id FROM calculated_metrics ORDER BY ticker, period_date, created_at DESC ) """)) await db.commit() logger.info(f"Duplicate cleanup completed") return { "status": "success", "message": "Duplicate records cleaned up successfully", "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Error cleaning up duplicates: {str(e)}") await db.rollback() raise HTTPException( status_code=500, detail=f"Failed to cleanup duplicates: {str(e)}" ) @router.get("/tickers") async def get_available_tickers(db: AsyncSession = Depends(get_db)) -> Dict[str, List[str]]: """ 사용 가능한 종목 목록을 반환합니다. """ try: # 재무 데이터가 있는 종목들 financial_tickers = await db.execute( select(distinct(FinancialData.ticker)).order_by(FinancialData.ticker) ) financial_list = [ticker for ticker, in financial_tickers.all()] # 주가 데이터가 있는 종목들 price_tickers = await db.execute( select(distinct(PriceData.ticker)).order_by(PriceData.ticker) ) price_list = [ticker for ticker, in price_tickers.all()] # 모든 종목들 all_tickers = await db.execute( select(Company.ticker).order_by(Company.ticker) ) all_list = [ticker for ticker, in all_tickers.all()] return { "tickers": list(set(financial_list + price_list + all_list)), "financial_tickers": financial_list, "price_tickers": price_list, "all_tickers": all_list } except Exception as e: logger.error(f"Error fetching available tickers: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to fetch available tickers: {str(e)}" ) # ========================== # ETF persisted data browsing # ========================== @router.get("/etf/snapshots") async def list_etf_snapshots( ticker: str | None = None, start_date: str | None = None, end_date: str | None = None, limit: int = 50, offset: int = 0, db: AsyncSession = Depends(get_db), ): try: from sqlalchemy import and_, desc q = select(ETFHoldingsSnapshot) conditions = [] if ticker: conditions.append(ETFHoldingsSnapshot.ticker == ticker.upper()) if start_date: from datetime import datetime, timezone try: sd = datetime.fromisoformat(start_date) if sd.tzinfo is None: sd = sd.replace(tzinfo=timezone.utc) conditions.append(ETFHoldingsSnapshot.snapshot_date >= sd) except Exception: pass if end_date: from datetime import datetime, timezone try: ed = datetime.fromisoformat(end_date) if ed.tzinfo is None: ed = ed.replace(tzinfo=timezone.utc) conditions.append(ETFHoldingsSnapshot.snapshot_date <= ed) except Exception: pass if conditions: from sqlalchemy import and_ as _and q = q.where(_and(*conditions)) q = q.order_by(desc(ETFHoldingsSnapshot.snapshot_date)).limit(limit).offset(offset) res = await db.execute(q) rows = res.scalars().all() # Count holdings per snapshot data = [] for s in rows: cnt_res = await db.execute(select(func.count(ETFHolding.id)).where(ETFHolding.snapshot_id == s.id)) hcount = cnt_res.scalar() or 0 data.append({ "id": str(s.id), "ticker": s.ticker, "snapshot_date": s.snapshot_date.isoformat() if s.snapshot_date else None, "source": s.source, "cik": s.cik, "filing_accession": s.filing_accession, "xml_url": s.xml_url, "holdings_count": hcount, }) return {"results": data, "count": len(data)} except Exception as e: logger.error(f"Error listing ETF snapshots: {e}") raise HTTPException(status_code=500, detail="Failed to list ETF snapshots") @router.get("/etf/snapshot/{snapshot_id}") async def get_etf_snapshot(snapshot_id: str, db: AsyncSession = Depends(get_db)): try: from uuid import UUID sid = UUID(snapshot_id) sres = await db.execute(select(ETFHoldingsSnapshot).where(ETFHoldingsSnapshot.id == sid)) snap = sres.scalar_one_or_none() if not snap: raise HTTPException(status_code=404, detail="Snapshot not found") hres = await db.execute(select(ETFHolding).where(ETFHolding.snapshot_id == sid)) holdings = [ { "name": h.name, "cusip": h.cusip, "ticker": h.ticker, "shares": h.shares, "value": h.value, "percentage": h.percentage, } for h in hres.scalars().all() ] return { "snapshot": { "id": str(snap.id), "ticker": snap.ticker, "snapshot_date": snap.snapshot_date.isoformat() if snap.snapshot_date else None, "source": snap.source, "cik": snap.cik, "filing_accession": snap.filing_accession, "xml_url": snap.xml_url, }, "holdings": holdings, "holdings_count": len(holdings), } except HTTPException: raise except Exception as e: logger.error(f"Error fetching ETF snapshot {snapshot_id}: {e}") raise HTTPException(status_code=500, detail="Failed to fetch ETF snapshot") # ============================== # Financial records browsing list # ============================== @router.get("/financial/records") async def list_financial_records( ticker: str | None = None, period_type: str | None = None, start_date: str | None = None, end_date: str | None = None, limit: int = 100, offset: int = 0, db: AsyncSession = Depends(get_db), ): try: from sqlalchemy import and_, desc q = select(FinancialData) conditions = [] if ticker: conditions.append(FinancialData.ticker == ticker.upper()) if period_type and period_type.lower() in ("quarterly", "annual"): conditions.append(FinancialData.period_type == period_type.lower()) from datetime import datetime, timezone if start_date: try: sd = datetime.fromisoformat(start_date) if sd.tzinfo is None: sd = sd.replace(tzinfo=timezone.utc) conditions.append(FinancialData.period_date >= sd) except Exception: pass if end_date: try: ed = datetime.fromisoformat(end_date) if ed.tzinfo is None: ed = ed.replace(tzinfo=timezone.utc) conditions.append(FinancialData.period_date <= ed) except Exception: pass if conditions: from sqlalchemy import and_ as _and q = q.where(_and(*conditions)) q = q.order_by(desc(FinancialData.period_date)).limit(limit).offset(offset) res = await db.execute(q) records = res.scalars().all() data = [] for r in records: data.append({ "ticker": r.ticker, "period_date": r.period_date.isoformat() if r.period_date else None, "period_type": r.period_type, "filing_type": r.filing_type, "revenue": r.revenue, "net_income": r.net_income, "total_assets": r.total_assets, "total_equity": r.total_equity, "shares_outstanding": r.shares_outstanding, "data_source": r.data_source, "is_estimated": r.is_estimated, }) return {"results": data, "count": len(data)} except Exception as e: logger.error(f"Error listing financial records: {e}") raise HTTPException(status_code=500, detail="Failed to list financial records")