#!/usr/bin/env python3 """ forecast_markets.py (updated) Fixes: - Drop rows with NaNs for each horizon before CV/training (avoids ValueError: Input contains NaN). - Safe fallback for TimeSeriesSplit when dataset is small. - Use DataFrames consistently to avoid sklearn feature-name warnings. Requirements: numpy pandas scikit-learn yfinance pymysql joblib """ from datetime import datetime, timedelta import os import json import joblib import warnings import numpy as np import pandas as pd import yfinance as yf import pymysql from sklearn.model_selection import TimeSeriesSplit from sklearn.ensemble import RandomForestRegressor from sklearn.linear_model import LogisticRegression from sklearn.metrics import mean_absolute_error, mean_squared_error, roc_auc_score, accuracy_score # ---------------- CONFIG ---------------- TICKER = "^GSPC" # change if you want FTSE, etc. VIX_TICKER = "^VIX" MODEL_DIR = os.path.expanduser("~/models") os.makedirs(MODEL_DIR, exist_ok=True) DB = { "host": "192.168.0.41", "user": "mawdz09", "password": "asaxPI1324", "database": "MarketForecasts" } FORECAST_TABLE = "market_forecasts" # Horizons: trading-day approx (21 ~ 1 month, 63 ~ 3 months) HORIZONS = {"1m": 21, "3m": 63} # ---------------- HELPERS ---------------- def get_price_data(tickers, start, end): raw = yf.download(tickers, start=start.strftime("%Y-%m-%d"), end=end.strftime("%Y-%m-%d"), progress=False, auto_adjust=False) if isinstance(raw, pd.Series): return raw.to_frame(name=str(tickers)) if isinstance(raw, pd.DataFrame): if 'Adj Close' in raw.columns: try: return raw['Adj Close'].copy() except Exception: pass if isinstance(raw.columns, pd.MultiIndex): lvl0 = raw.columns.get_level_values(0) lvl1 = raw.columns.get_level_values(1) if 'Adj Close' in lvl0: return raw.xs('Adj Close', axis=1, level=0).copy() if 'Adj Close' in lvl1: return raw.xs('Adj Close', axis=1, level=1).copy() return raw.copy() raise RuntimeError("Unexpected yfinance result structure; examine `raw` output manually.") def ensure_ticker_columns(df, tickers): def col_name_repr(c): if isinstance(c, tuple): return " ".join([str(x) for x in c]) return str(c) cols = [col_name_repr(c) for c in df.columns] mapping = {} for t in tickers: if t in cols: mapping[t] = cols.index(t) continue t_simpl = ''.join(ch for ch in t if ch.isalnum()) found = None for i, c in enumerate(cols): if t_simpl and t_simpl in ''.join(ch for ch in c if ch.isalnum()): found = i break if t in c: found = i break if found is not None: mapping[t] = found else: raise KeyError(f"Couldn't locate ticker '{t}' in downloaded data columns: {cols[:10]} ...") out = pd.DataFrame(index=df.index) for t in tickers: out[t] = df.iloc[:, mapping[t]] return out def rsi(series, n=14): delta = series.diff() up = delta.clip(lower=0).rolling(n).mean() down = -delta.clip(upper=0).rolling(n).mean() rs = up / (down + 1e-12) return 100 - (100 / (1 + rs)) # ---------------- MAIN ---------------- def main(): warnings.filterwarnings("ignore", category=FutureWarning) end = datetime.today() start = end - timedelta(days=3650) # ~10 years history print("Downloading price data...") raw_prices = get_price_data([TICKER, VIX_TICKER], start, end) try: prices = ensure_ticker_columns(raw_prices, [TICKER, VIX_TICKER]) except KeyError as e: print("Error matching tickers to downloaded columns:", e) print("Downloaded columns (first 20):", list(raw_prices.columns)[:20]) raise prices = prices.dropna() print("Price columns:", prices.columns.tolist()) if prices.shape[0] < 300: print("Warning: short history (rows):", prices.shape[0]) # Build feature frame idx = prices[TICKER].copy() frame = pd.DataFrame(index=idx.index) frame['close'] = idx frame['ret_1d'] = frame['close'].pct_change(1) frame['ret_5d'] = frame['close'].pct_change(5) frame['ret_21d'] = frame['close'].pct_change(21) frame['vol_21'] = frame['ret_1d'].rolling(21).std() * (252 ** 0.5) frame['ma50'] = frame['close'].rolling(50).mean() frame['ma200'] = frame['close'].rolling(200).mean() frame['ma_spread'] = (frame['ma50'] - frame['ma200']) / frame['ma200'] frame['maxdd_90'] = frame['close'].rolling(90).apply(lambda x: (x.max() - x.min()) / x.max() if x.max() != 0 else 0) frame['rsi14'] = rsi(frame['close'], 14) frame['vix'] = prices[VIX_TICKER] frame = frame.copy() if frame.empty: raise RuntimeError("No rows after feature construction; check price data.") # Labels for horizons for label, horizon in HORIZONS.items(): frame[f'future_close_{label}'] = frame['close'].shift(-horizon) frame[f'target_pct_{label}'] = (frame[f'future_close_{label}'] - frame['close']) / frame['close'] frame[f'target_up_{label}'] = (frame[f'target_pct_{label}'] > 0).astype(int) # Keep at least the shortest-horizon label rows (we'll drop per-horizon NaNs later) shortest_label = list(HORIZONS.keys())[0] frame = frame.dropna(subset=[f'target_pct_{shortest_label}']) feature_cols = ['ret_1d', 'ret_5d', 'ret_21d', 'vol_21', 'ma_spread', 'maxdd_90', 'rsi14', 'vix'] results = {} for label, horizon in HORIZONS.items(): print(f"\nPreparing dataset for horizon {label} ({horizon} days)...") # Build horizon-specific dataset and drop any rows with NaNs in features or the labels for this horizon cols_needed = feature_cols + [f'target_pct_{label}', f'target_up_{label}'] df_h = frame[cols_needed].dropna() if df_h.shape[0] < 30: print(f" Not enough rows ({df_h.shape[0]}) to reliably train for {label}. Skipping.") continue X_h = df_h[feature_cols] y_reg = df_h[f'target_pct_{label}'] y_cls = df_h[f'target_up_{label}'] print(f" rows available: {len(X_h)}") # choose n_splits for TimeSeriesSplit reasonably max_splits = 5 n_splits = min(max_splits, max(2, len(X_h) // 20)) # roughly get at least ~20 rows per split if possible if n_splits < 2: n_splits = 2 tscv = TimeSeriesSplit(n_splits=n_splits) reg_mae, reg_rmse, cls_auc, cls_acc = [], [], [], [] for train_idx, test_idx in tscv.split(X_h): Xtr, Xte = X_h.iloc[train_idx], X_h.iloc[test_idx] ytr_reg, yte_reg = y_reg.iloc[train_idx], y_reg.iloc[test_idx] ytr_cls, yte_cls = y_cls.iloc[train_idx], y_cls.iloc[test_idx] # train regressor reg = RandomForestRegressor(n_estimators=100, random_state=42) reg.fit(Xtr, ytr_reg) pred_reg = pd.Series(reg.predict(Xte), index=Xte.index) reg_mae.append(mean_absolute_error(yte_reg, pred_reg)) mse_val = mean_squared_error(yte_reg, pred_reg) reg_rmse.append(float(np.sqrt(mse_val))) # train classifier (logistic) cls = LogisticRegression(max_iter=1000) cls.fit(Xtr, ytr_cls) prob = pd.Series(cls.predict_proba(Xte)[:, 1], index=Xte.index) try: cls_auc.append(roc_auc_score(yte_cls, prob)) except Exception: cls_auc.append(0.0) cls_acc.append(accuracy_score(yte_cls, (prob > 0.5).astype(int))) results[label] = { "reg_mae_cv": float(np.mean(reg_mae)) if reg_mae else None, "reg_rmse_cv": float(np.mean(reg_rmse)) if reg_rmse else None, "cls_auc_cv": float(np.mean(cls_auc)) if cls_auc else None, "cls_acc_cv": float(np.mean(cls_acc)) if cls_acc else None } print("CV reg MAE:", results[label]["reg_mae_cv"], "RMSE:", results[label]["reg_rmse_cv"]) print("CV cls AUC:", results[label]["cls_auc_cv"], "Acc:", results[label]["cls_acc_cv"]) # Train final models on full horizon-cleaned dataset final_reg = RandomForestRegressor(n_estimators=200, random_state=42) final_reg.fit(X_h, y_reg) final_cls = LogisticRegression(max_iter=1000) final_cls.fit(X_h, y_cls) joblib.dump({"model": final_reg, "features": feature_cols, "horizon": horizon}, os.path.join(MODEL_DIR, f"reg_{label}.joblib")) joblib.dump({"model": final_cls, "features": feature_cols, "horizon": horizon}, os.path.join(MODEL_DIR, f"cls_{label}.joblib")) # Latest forecast uses the last available row in df_h (which is safe for this horizon) latest_row = df_h.iloc[-1] X_latest = latest_row[feature_cols].to_frame().T pred_pct = float(final_reg.predict(X_latest)[0]) prob_up = float(final_cls.predict_proba(X_latest)[0, 1]) results[label].update({ "pred_pct": pred_pct, "prob_up": prob_up, "latest_date": df_h.index[-1].strftime("%Y-%m-%d"), "rows_used": int(df_h.shape[0]) }) # ---------------- Write to DB ---------------- print("\nSaving forecasts to DB...") conn = pymysql.connect(host=DB['host'], user=DB['user'], password=DB['password'], charset='utf8mb4', autocommit=True) cur = conn.cursor() try: cur.execute(f"CREATE DATABASE IF NOT EXISTS `{DB['database']}` DEFAULT CHARACTER SET utf8mb4;") except Exception as e: print("DB creation warning:", e) cur.close() conn.close() conn = pymysql.connect(host=DB['host'], user=DB['user'], password=DB['password'], database=DB['database'], charset='utf8mb4') cur = conn.cursor() create_table_sql = f""" CREATE TABLE IF NOT EXISTS `{FORECAST_TABLE}` ( id INT AUTO_INCREMENT PRIMARY KEY, ts DATETIME NOT NULL, ticker VARCHAR(32) NOT NULL, horizon_label VARCHAR(8) NOT NULL, horizon_days INT NOT NULL, pred_pct FLOAT, prob_up FLOAT, model VARCHAR(128), features_json TEXT, actual_pct FLOAT, realized_up TINYINT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """ cur.execute(create_table_sql) # insert latest for each horizon we have results for for label, horizon in HORIZONS.items(): if label not in results or results[label] is None: continue r = results[label] ins_sql = f"INSERT INTO `{FORECAST_TABLE}` (ts, ticker, horizon_label, horizon_days, pred_pct, prob_up, model, features_json) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)" # use last row from frame if available; else current datetime try: ts_val = pd.to_datetime(r.get("latest_date")).to_pydatetime() except Exception: ts_val = datetime.now() features_snapshot = json.dumps({f: float(frame.iloc[-1][f]) for f in feature_cols}) cur.execute(ins_sql, (ts_val, TICKER, label, horizon, r['pred_pct'], r['prob_up'], f"rf+logreg_v1_{label}", features_snapshot)) conn.commit() cur.close() conn.close() print("\nForecasts written. Summary:") for label in results: print(label, results[label]) if __name__ == "__main__": main()