#!/usr/bin/env python3 """ score_forecasts.py - Finds forecasts in MarketForecasts.market_forecasts with actual_pct IS NULL and where (ts + horizon_days) trading days have passed. - Uses yfinance to load index prices around the forecast ts, finds the index price at ts (next trading day >= ts) and at ts + horizon_days (index position + horizon_days), computes actual_pct, realized_up and updates the DB row. - Computes aggregate metrics (MAE, RMSE, accuracy, AUC if possible) for updated rows and stores them in forecast_performance table. Requirements: numpy pandas yfinance pymysql scikit-learn """ from datetime import datetime, timedelta import json import os import sys import numpy as np import pandas as pd import yfinance as yf import pymysql from sklearn.metrics import mean_absolute_error, mean_squared_error, accuracy_score, roc_auc_score, brier_score_loss # ---------- CONFIG ---------- DB = { "host": "192.168.0.41", "user": "mawdz09", "password": "asaxPI1324", "database": "MarketForecasts" } FORECAST_TABLE = "market_forecasts" PERF_TABLE = "forecast_performance" # ---------- HELPERS ---------- def fetch_price_series(ticker, start_dt, end_dt): """ Fetch adjusted-close series for a single ticker between start_dt and end_dt. Returns a pandas Series indexed by Timestamp (sorted ascending). """ raw = yf.download(ticker, start=start_dt.strftime("%Y-%m-%d"), end=(end_dt + timedelta(days=1)).strftime("%Y-%m-%d"), progress=False, auto_adjust=False) # Extract Adj Close robustly: if isinstance(raw, pd.Series): # single-col Series series = raw elif isinstance(raw, pd.DataFrame): if 'Adj Close' in raw.columns: series = raw['Adj Close'] elif ticker in raw.columns: series = raw[ticker] else: # If multiindex or some other layout, flatten and try to find a numeric column # Prefer last column as fallback for col in raw.columns[::-1]: try: s = pd.to_numeric(raw[col], errors='coerce') if not s.dropna().empty: series = s break except Exception: continue else: raise RuntimeError(f"Couldn't find price column for {ticker} in fetched frame; columns: {raw.columns}") else: raise RuntimeError("Unexpected yfinance return type") # ensure datetime index and sorted series.index = pd.to_datetime(series.index) series = series.sort_index() return series def find_price_by_offset(prices, base_dt, horizon_days): """ Given a prices Series indexed by trading datetimes, find: - pos_start = first index >= base_dt (bfill) - pos_target = pos_start + horizon_days (index offset in trading days) Return (price_start, price_target) or (None, None) if not possible. """ if prices.empty: return None, None # find position of first trading day >= base_dt # get integer location using get_indexer with method='bfill' pos = prices.index.get_indexer([pd.to_datetime(base_dt)], method='bfill') if pos is None or len(pos) == 0: return None, None pos0 = int(pos[0]) if pos0 < 0 or pos0 >= len(prices): return None, None pos_target = pos0 + int(horizon_days) if pos_target >= len(prices): # horizon target not yet available in series return None, None price_start = float(prices.iloc[pos0].item()) price_target = float(prices.iloc[pos0].item()) return price_start, price_target # ---------- MAIN ---------- def main(): # connect DB conn = pymysql.connect(host=DB['host'], user=DB['user'], password=DB['password'], charset='utf8mb4', autocommit=True) cur = conn.cursor() # ensure DB exists (should already) cur.execute(f"CREATE DATABASE IF NOT EXISTS `{DB['database']}` DEFAULT CHARACTER SET utf8mb4;") cur.close() conn.close() conn = pymysql.connect(host=DB['host'], user=DB['user'], password=DB['password'], database=DB['database'], charset='utf8mb4') cur = conn.cursor(pymysql.cursors.DictCursor) # fetch forecasts needing scoring cur.execute(f"SELECT id, ts, ticker, horizon_label, horizon_days, pred_pct, prob_up FROM {FORECAST_TABLE} WHERE actual_pct IS NULL ORDER BY ts ASC") rows = cur.fetchall() if not rows: print("No unscored forecasts found.") cur.close() conn.close() return updated_ids = [] preds = [] actuals = [] probs = [] realized = [] for r in rows: fid = r['id'] ts = r['ts'] # this is datetime object courtesy of DictCursor ticker = r['ticker'] horizon_days = int(r['horizon_days']) pred_pct = r['pred_pct'] prob_up = r['prob_up'] # fetch price series covering ts through ts + horizon_days + a buffer start_dt = pd.to_datetime(ts) - pd.Timedelta(days=7) end_dt = pd.to_datetime(ts) + pd.Timedelta(days=horizon_days + 14) try: prices = fetch_price_series(ticker, start_dt, end_dt) except Exception as ex: print(f" Skipping id {fid}: price fetch error for {ticker}: {ex}") continue price_start, price_target = find_price_by_offset(prices, ts, horizon_days) if price_start is None or price_target is None: # not yet available or cannot align trading days print(f" Skipping id {fid}: insufficient price range to compute horizon (ts {ts}, horizon {horizon_days})") continue actual_pct = (price_target - price_start) / price_start realized_up = 1 if actual_pct > 0 else 0 # update DB row upd_sql = f"UPDATE {FORECAST_TABLE} SET actual_pct=%s, realized_up=%s WHERE id=%s" cur.execute(upd_sql, (float(actual_pct), int(realized_up), int(fid))) conn.commit() print(f" Scored id {fid}: pred {pred_pct:.4f}, actual {actual_pct:.4f}, prob_up {prob_up:.3f} -> realized {realized_up}") # accumulate updated_ids.append(fid) preds.append(float(pred_pct) if pred_pct is not None else np.nan) actuals.append(float(actual_pct)) probs.append(float(prob_up) if prob_up is not None else np.nan) realized.append(int(realized_up)) # compute aggregate metrics if we updated at least one if updated_ids: preds = np.array(preds, dtype=float) actuals = np.array(actuals, dtype=float) probs = np.array(probs, dtype=float) realized = np.array(realized, dtype=int) # regression metrics mae = float(mean_absolute_error(actuals, preds)) rmse = float(np.sqrt(mean_squared_error(actuals, preds))) # classification metrics acc = float(accuracy_score(realized, (probs > 0.5).astype(int))) if len(probs) > 0 else None auc = None try: if len(np.unique(realized)) > 1 and len(probs) >= 10: auc = float(roc_auc_score(realized, probs)) except Exception: auc = None brier = float(brier_score_loss(realized, probs)) if len(probs) > 0 else None # print summary print("\n--- Scoring summary ---") print("Updated rows:", len(updated_ids)) print(f"MAE: {mae:.6f}, RMSE: {rmse:.6f}") print(f"Accuracy (prob>0.5): {acc}, AUC: {auc}, Brier: {brier}") # store performance snapshot into DB table forecast_performance cur.execute(f""" CREATE TABLE IF NOT EXISTS `{PERF_TABLE}` ( id INT AUTO_INCREMENT PRIMARY KEY, ts DATETIME NOT NULL, n_updated INT, mae FLOAT, rmse FLOAT, accuracy FLOAT, auc FLOAT, brier FLOAT, ids_json TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """) import json from datetime import datetime, timezone # Example forecast scoring results — replace with your actual logic updated_ids = [1] # list of forecast IDs that were successfully scored mae = 0.0001 # mean absolute error rmse = 0.0001 # root mean squared error accuracy = 1.0 # accuracy score (e.g. prob > 0.5) auc = None # area under curve (can be None if not computed) brier = 0.08693 # Brier score import pymysql from datetime import datetime, timezone import json # Replace with your actual credentials DB_HOST = "192.168.0.41" DB_PORT = 3306 DB_USER = "mawdz09" DB_PASS = "asaxPI1324" DB_NAME = "MarketForecasts" conn = pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS, database=DB_NAME, charset="utf8mb4", cursorclass=pymysql.cursors.Cursor ) cur = conn.cursor() if updated_ids: cur.execute( f"INSERT INTO {PERF_TABLE} (ts, n_updated, mae, rmse, accuracy, auc, brier, ids_json) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", ( datetime.now(timezone.utc), len(updated_ids), mae, rmse, accuracy, auc if auc is not None else 0.0, brier, json.dumps(updated_ids) ) ) conn.commit() else: print("No forecasts were scored (no completed horizons).") cur.close() conn.close() if __name__ == "__main__": main()