Spaces:
Sleeping
Sleeping
File size: 8,384 Bytes
d40d32b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | 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())
|