import asyncio import sys import os import time from pathlib import Path from dotenv import load_dotenv # Add parent directory to sys.path and load dotenv safely sys.path.insert(0, str(Path(__file__).parent.parent)) from app.services.database_service import DatabaseService dotenv_path = Path(__file__).parent.parent / ".env" load_dotenv(dotenv_path=dotenv_path) async def run_diagnostics(): url = os.getenv("TURSO_URL") token = os.getenv("TURSO_TOKEN") if not url or not token: print("Error: TURSO_URL and TURSO_TOKEN environment variables must be set in .env") sys.exit(1) db = DatabaseService(url=url, token=token) await db.initialize() try: print("=" * 60) print("DATABASE DIAGNOSTICS REPORT") print("=" * 60) # 1. Stock Coverage Breakdown print("\n[1] Stock Coverage Breakdown") coverage_sql = """ SELECT s.country, COUNT(CASE WHEN s.trading_status = 'active' THEN 1 END) as total_active, COUNT(CASE WHEN s.trading_status = 'inactive' THEN 1 END) as total_inactive, COUNT(DISTINCT CASE WHEN s.trading_status = 'active' AND f.symbol IS NOT NULL THEN s.yf_symbol END) as forecasted_active FROM ( SELECT country, COALESCE(trading_status, 'active') as trading_status, TRIM(REPLACE(symbol, 'NZX:', '')) || COALESCE(yfinance_suffix, '') as yf_symbol FROM stocks ) s LEFT JOIN ( SELECT DISTINCT symbol FROM forecasts ) f ON s.yf_symbol = f.symbol GROUP BY s.country ORDER BY s.country """ rows = await db._execute(coverage_sql) print(f"{'Country':<10} | {'Active':<8} | {'Inactive':<8} | {'Forecasted':<10} | {'Coverage':<8}") print("-" * 60) for r in rows: active = r['total_active'] forecasted = r['forecasted_active'] pct = (forecasted / active * 100) if active > 0 else 0.0 print(f"{r['country'].upper():<10} | {active:<8} | {r['total_inactive']:<8} | {forecasted:<10} | {pct:.2f}%") # 2. Freshness & Staleness print("\n[2] Forecast Freshness & Staleness (Last 24 Hours)") now = int(time.time()) threshold = now - (24 * 3600) freshness_sql = """ SELECT COUNT(CASE WHEN created_at >= ? THEN 1 END) as fresh_count, COUNT(CASE WHEN created_at < ? THEN 1 END) as stale_count FROM forecasts """ fresh_rows = await db._execute(freshness_sql, [threshold, threshold]) print(f"Fresh Forecasts (Last 24h): {fresh_rows[0]['fresh_count']}") print(f"Stale Forecasts (> 24h): {fresh_rows[0]['stale_count']}") print("\n[2b] 5 Oldest Forecasts in Database") oldest_sql = """ SELECT symbol, horizon_days, created_at FROM forecasts ORDER BY created_at ASC LIMIT 5 """ old_rows = await db._execute(oldest_sql) for idx, r in enumerate(old_rows, 1): created_str = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(r['created_at'])) print(f" {idx}. {r['symbol']:<10} ({r['horizon_days']}D) - Created: {created_str} UTC") # 3. High Market Cap Missing Forecasts print("\n[3] Top 5 Missing Forecasts per Country (by Market Cap)") countries_res = await db._execute("SELECT DISTINCT country FROM stocks ORDER BY country") countries = [r['country'] for r in countries_res if r['country']] for country in countries: missing_sql = """ SELECT s.symbol, s.name, s.market_cap FROM stocks s LEFT JOIN ( SELECT DISTINCT symbol FROM forecasts ) f ON (TRIM(REPLACE(s.symbol, 'NZX:', '')) || COALESCE(s.yfinance_suffix, '')) = f.symbol WHERE s.country = ? AND (s.trading_status = 'active' OR s.trading_status IS NULL) AND (s.yfinance_available = 1 OR s.yfinance_available IS NULL) AND f.symbol IS NULL ORDER BY s.market_cap DESC NULLS LAST LIMIT 5 """ m_rows = await db._execute(missing_sql, [country]) print(f" Country: {country.upper()}") for r in m_rows: mcap_str = f"${r['market_cap'] / 1e9:.2f}B" if r['market_cap'] else "N/A" print(f" - {r['symbol']:<8} | {r['name'][:30]:<30} | Market Cap: {mcap_str}") # 4. Pipeline Integrity - Mismatched Horizons print("\n[4] Pipeline Integrity: Mismatched Horizon Counts (< 4)") mismatched_sql = """ SELECT symbol, COUNT(DISTINCT horizon_days) as horizon_count, GROUP_CONCAT(horizon_days) as horizons FROM forecasts GROUP BY symbol HAVING horizon_count < 4 """ mismatched_rows = await db._execute(mismatched_sql) if mismatched_rows: # Map back to stock details in Python to prevent slow SQL joins sym_map = {} for row in mismatched_rows[:10]: yf_sym = row['symbol'] base = yf_sym for suff in [".L", ".TO", ".AX", ".NZ", ".IR"]: if yf_sym.endswith(suff): base = yf_sym[:-len(suff)] break sym_map[base] = row placeholders = ", ".join(["?"] * len(sym_map)) stocks_sql = f"SELECT symbol, country FROM stocks WHERE symbol IN ({placeholders})" stock_rows = await db._execute(stocks_sql, list(sym_map.keys())) stock_info = {r['symbol']: r for r in stock_rows} print(f" Found {len(mismatched_rows)} symbols with mismatched horizons (showing top 10):") for row in mismatched_rows[:10]: yf_sym = row['symbol'] base = yf_sym for suff in [".L", ".TO", ".AX", ".NZ", ".IR"]: if yf_sym.endswith(suff): base = yf_sym[:-len(suff)] break country = stock_info.get(base, {}).get('country', 'us').upper() print(f" - {yf_sym:<10} ({country}) | Horizons present: {row['horizons']}") else: print(" No mismatched horizon counts found. All forecasted stocks have exactly 4 horizons.") # 5. Pipeline Integrity - Out of Sync Horizons print("\n[5] Pipeline Integrity: Out of Sync Horizons (> 60s difference)") sync_sql = """ SELECT symbol, (MAX(created_at) - MIN(created_at)) as time_diff FROM forecasts GROUP BY symbol HAVING time_diff > 60 """ sync_rows = await db._execute(sync_sql) if sync_rows: sym_map = {} for row in sync_rows[:10]: yf_sym = row['symbol'] base = yf_sym for suff in [".L", ".TO", ".AX", ".NZ", ".IR"]: if yf_sym.endswith(suff): base = yf_sym[:-len(suff)] break sym_map[base] = row placeholders = ", ".join(["?"] * len(sym_map)) stocks_sql = f"SELECT symbol, country FROM stocks WHERE symbol IN ({placeholders})" stock_rows = await db._execute(stocks_sql, list(sym_map.keys())) stock_info = {r['symbol']: r for r in stock_rows} print(f" Found {len(sync_rows)} symbols out of sync (showing top 10):") for row in sync_rows[:10]: yf_sym = row['symbol'] base = yf_sym for suff in [".L", ".TO", ".AX", ".NZ", ".IR"]: if yf_sym.endswith(suff): base = yf_sym[:-len(suff)] break country = stock_info.get(base, {}).get('country', 'us').upper() print(f" - {yf_sym:<10} ({country}) | Max creation delay: {row['time_diff']}s") else: print(" No out of sync horizons found. All forecasts generated within 60 seconds of each other.") print("=" * 60) finally: await db.close() if __name__ == "__main__": asyncio.run(run_diagnostics())