Spaces:
Sleeping
Sleeping
| 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()) | |