Spaces:
Sleeping
Sleeping
File size: 5,221 Bytes
99938b8 | 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 | """
Free-tier walk-forward accuracy seed.
Generates a small set of historical forecast_snapshots with honest statistical
p50 forecasts and true closes so /accuracy is non-empty without paid APIs.
Usage (from backend/ with venv):
python scripts/seed_accuracy_backtest.py --symbols AAPL,MSFT,GOOGL --horizons 1,5,20 --windows 30
Keeps Yahoo + Turso writes tiny (default ~ a few hundred rows).
"""
from __future__ import annotations
import argparse
import asyncio
import os
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from dotenv import load_dotenv
load_dotenv()
from app.services.accuracy_service import close_on_or_before, compute_errors
from app.services.data_service import StockDataService
from app.services.database_service import DatabaseService
from app.services.timesfm_service import TimesFMService
async def seed(
symbols: list[str],
horizons: list[int],
windows: int,
delay: float,
) -> dict:
db = DatabaseService(url=os.getenv("TURSO_URL"), token=os.getenv("TURSO_TOKEN"))
data = StockDataService()
model = TimesFMService()
# Free-tier seed: skip heavy model load; statistical path is honest and labeled
model.available = False
model.model = None
written = 0
evaluated = 0
try:
await db.initialize()
for symbol in symbols:
hist = await data.get_stock_data(symbol, period="2y")
if hist is None or len(hist) < 80:
print(f"[skip] {symbol}: insufficient history")
continue
closes = [float(x) for x in hist["Close"].tolist()]
# Walk back from end: use windows ending before last bar
max_h = max(horizons)
end_idx = len(closes) - 1
start_idx = max(60, end_idx - windows - max_h)
for i in range(start_idx, end_idx - max_h):
context = closes[: i + 1]
entry = context[-1]
# Forecast date ≈ bar timestamp
ts = hist.index[i]
forecast_date = int(ts.timestamp()) // 86400 * 86400
for h in horizons:
target_i = i + h
if target_i >= len(closes):
continue
pred = await model.predict(context, horizon=h)
p50 = float(pred["quantiles"]["p50"][-1])
# Insert snapshot as if made on forecast_date
await db._execute(
"""
INSERT OR IGNORE INTO forecast_snapshots
(symbol, country, exchange, forecast_date, horizon_days, target_date,
p50_price, entry_price, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
symbol.upper(),
"us",
"",
forecast_date,
h,
forecast_date + h * 86400,
p50,
entry,
forecast_date,
],
)
written += 1
# Evaluate immediately with known actual
actual = float(closes[target_i])
pct_err, direction_ok = compute_errors(entry, p50, actual)
await db._execute(
"""
UPDATE forecast_snapshots
SET actual_price = ?, percentage_error = ?, direction_correct = ?,
evaluated_at = ?
WHERE symbol = ? AND horizon_days = ? AND forecast_date = ?
AND evaluated_at IS NULL
""",
[
actual,
pct_err,
direction_ok,
int(time.time()),
symbol.upper(),
h,
forecast_date,
],
)
evaluated += 1
print(f"[ok] {symbol}")
await asyncio.sleep(delay)
finally:
await db.close()
return {"written": written, "evaluated": evaluated}
def main():
p = argparse.ArgumentParser()
p.add_argument("--symbols", default="AAPL,MSFT,GOOGL,AMZN,META,NVDA,JPM,XOM,JNJ,WMT")
p.add_argument("--horizons", default="1,5,20,60")
p.add_argument("--windows", type=int, default=40, help="Lookback windows per symbol (keep small)")
p.add_argument("--delay", type=float, default=0.3)
args = p.parse_args()
symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]
horizons = [int(x) for x in args.horizons.split(",") if x.strip()]
# Cap to protect free Turso write quota
windows = min(args.windows, 60)
result = asyncio.run(seed(symbols, horizons, windows, args.delay))
print(result)
if __name__ == "__main__":
main()
|