""" Pattern Lab — Historical backtest engine for pattern discovery. Workflow: 1. build_historical_context(date, tickers) — pull prices + technicals via yfinance 2. run_ai_backtest(context, hint, horizon) — GPT-4o acts as analyst on that past date 3. evaluate_outcomes(run) — fetch actual prices at T+horizon, score each idea """ import json import logging from datetime import datetime, timedelta from typing import Optional import numpy as np import pandas as pd import yfinance as yf _log = logging.getLogger(__name__) # ── Context builder ──────────────────────────────────────────────────────────── def _fetch_ticker(ticker: str, start: str, end: str) -> Optional[pd.DataFrame]: try: df = yf.download(ticker, start=start, end=end, progress=False, auto_adjust=True) if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) return df if not df.empty else None except Exception as e: _log.warning(f"[PatternLab] yfinance error for {ticker}: {e}") return None def _rsi(closes: pd.Series, period: int = 14) -> Optional[float]: if len(closes) < period + 1: return None delta = closes.diff() gain = delta.clip(lower=0).rolling(period).mean() loss = (-delta.clip(upper=0)).rolling(period).mean() rs = gain / loss val = 100 - (100 / (1 + rs.iloc[-1])) return round(float(val), 1) if not np.isnan(val) else None def build_historical_context(analysis_date: str, tickers: list) -> dict: dt = datetime.strptime(analysis_date, "%Y-%m-%d") hist_start = (dt - timedelta(days=400)).strftime("%Y-%m-%d") hist_end = (dt + timedelta(days=3)).strftime("%Y-%m-%d") context: dict = {"analysis_date": analysis_date, "assets": {}} all_tickers = list(tickers) if "^VIX" not in all_tickers: all_tickers.append("^VIX") for ticker in all_tickers: df = _fetch_ticker(ticker, hist_start, hist_end) if df is None or "Close" not in df.columns: context["assets"][ticker] = {"error": "no data"} continue df.index = pd.to_datetime(df.index).tz_localize(None) sub = df[df.index.normalize() <= dt].copy() if sub.empty: context["assets"][ticker] = {"error": "no data before date"} continue closes = sub["Close"].dropna() price = float(closes.iloc[-1]) def ret(days: int) -> Optional[float]: past = sub[sub.index.normalize() <= dt - timedelta(days=days)] if past.empty: return None p0 = float(past["Close"].dropna().iloc[-1]) return round((price / p0 - 1) * 100, 2) if p0 else None ma50 = round(float(closes.tail(50).mean()), 2) if len(closes) >= 50 else None ma200 = round(float(closes.tail(200).mean()), 2) if len(closes) >= 200 else None context["assets"][ticker] = { "price": round(price, 4), "ret_1w": ret(5), "ret_1m": ret(21), "ret_3m": ret(63), "ret_ytd": ret(int((dt - datetime(dt.year, 1, 1)).days)), "rsi_14": _rsi(closes.tail(60)), "ma50": ma50, "ma200": ma200, "pct_from_ma200": round((price / ma200 - 1) * 100, 2) if ma200 else None, } return context # ── AI analysis ──────────────────────────────────────────────────────────────── async def run_ai_backtest(context: dict, theme_hint: str, horizon_days: int, openai_key: str) -> dict: from openai import AsyncOpenAI client = AsyncOpenAI(api_key=openai_key) analysis_date = context["analysis_date"] assets = context.get("assets", {}) lines = [] for ticker, d in assets.items(): if "error" in d: continue line = f" {ticker}: {d.get('price', '?')}" if d.get("ret_1w") is not None: line += f" 1W:{d['ret_1w']:+.1f}%" if d.get("ret_1m") is not None: line += f" 1M:{d['ret_1m']:+.1f}%" if d.get("ret_3m") is not None: line += f" 3M:{d['ret_3m']:+.1f}%" if d.get("rsi_14") is not None: line += f" RSI:{d['rsi_14']:.0f}" if d.get("pct_from_ma200") is not None: line += f" vsMA200:{d['pct_from_ma200']:+.1f}%" lines.append(line) market_block = "\n".join(lines) if lines else "No market data available." prompt = f"""You are a senior macro/options trader. Today is {analysis_date}. You have full knowledge of what was happening in markets and geopolitics on this specific date. ## REAL MARKET DATA — {analysis_date} {market_block} ## CONTEXT / THEME {theme_hint} ## INSTRUCTIONS Using the market data above AND your knowledge of the historical context on {analysis_date}: - Identify 2 to 4 high-conviction trade patterns that were present at this time. - For each pattern, recommend the best options/futures strategy. - The investment horizon is {horizon_days} days from {analysis_date}. For each pattern you MUST specify: - The primary underlying ticker (must be a real, liquid, yfinance-compatible symbol) - The expected percentage move in the UNDERLYING over {horizon_days} days - The expected direction: "up", "down", or "any" (for volatility plays) - A confidence score 0–100 Return ONLY this valid JSON (no markdown, no explanation): {{ "patterns": [ {{ "name": "Short descriptive pattern name", "description": "What you observed and why this matters", "category": "géopolitique|macro_monétaire|technique|commodités_supply|risk_off|flux_saisonnier|géo_économique|crédit_stress", "signal_direction": "bullish|bearish|volatility|neutral", "underlying": "TICKER", "strategy": "e.g. long put, long call, straddle, call spread", "expected_move_pct": 12.5, "expected_direction": "up|down|any", "horizon_days": {horizon_days}, "confidence": 72, "rationale": "Concise explanation of the trade thesis at this historical moment" }} ] }}""" try: resp = await client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], temperature=0.2, response_format={"type": "json_object"}, timeout=60, ) raw = resp.choices[0].message.content or "{}" return json.loads(raw) except Exception as e: _log.error(f"[PatternLab] AI backtest failed: {e}") return {"patterns": [], "error": str(e)} # ── Outcome evaluator ────────────────────────────────────────────────────────── def evaluate_outcomes(run: dict) -> list: analysis_date = run["analysis_date"] horizon_days = int(run["horizon_days"]) ai_result = json.loads(run.get("ai_result") or "{}") patterns = ai_result.get("patterns", []) dt = datetime.strptime(analysis_date, "%Y-%m-%d") eval_dt = dt + timedelta(days=horizon_days) # Don't evaluate if eval date is in the future if eval_dt.date() >= datetime.utcnow().date(): return [{"error": f"Evaluation date {eval_dt.date()} is in the future"}] fetch_start = (dt - timedelta(days=5)).strftime("%Y-%m-%d") fetch_end = (eval_dt + timedelta(days=5)).strftime("%Y-%m-%d") outcomes = [] for pat in patterns: ticker = pat.get("underlying", "") if not ticker: continue df = _fetch_ticker(ticker, fetch_start, fetch_end) if df is None or "Close" not in df.columns: outcomes.append({"pattern_name": pat.get("name"), "underlying": ticker, "error": "no data"}) continue df.index = pd.to_datetime(df.index).tz_localize(None) entry_sub = df[df.index.normalize() <= dt] exit_sub = df[df.index.normalize() <= eval_dt] if entry_sub.empty or exit_sub.empty: outcomes.append({"pattern_name": pat.get("name"), "underlying": ticker, "error": "insufficient history"}) continue entry_price = float(entry_sub["Close"].dropna().iloc[-1]) exit_price = float(exit_sub["Close"].dropna().iloc[-1]) actual_move = round((exit_price / entry_price - 1) * 100, 2) expected_dir = pat.get("expected_direction", "any") expected_move = float(pat.get("expected_move_pct", 0)) threshold = max(expected_move * 0.5, 3.0) # at least 50% of expected, min 3% if expected_dir == "up": hit = actual_move >= threshold elif expected_dir == "down": hit = actual_move <= -threshold else: # any / volatility hit = abs(actual_move) >= threshold outcomes.append({ "pattern_name": pat.get("name"), "underlying": ticker, "strategy": pat.get("strategy", ""), "signal_direction": pat.get("signal_direction", ""), "expected_direction": expected_dir, "expected_move_pct": expected_move, "actual_move_pct": actual_move, "entry_price": round(entry_price, 4), "exit_price": round(exit_price, 4), "entry_date": analysis_date, "exit_date": eval_dt.strftime("%Y-%m-%d"), "hit": hit, "confidence": pat.get("confidence", 0), }) return outcomes