#!/usr/bin/env python3 """ 1-Month Stock Price & Gain Predictor — Full Alpha Vantage Intelligence ======================================================================= Uses EVERY available Alpha Vantage API to build the most comprehensive 30-day price forecast possible for a given ticker. Data Sources (16 API endpoints): 1. TIME_SERIES_DAILY_ADJUSTED — 5yr OHLCV history 2. GLOBAL_QUOTE — latest realtime price 3. OVERVIEW — fundamentals (P/E, EPS, margins, etc.) 4. INCOME_STATEMENT — revenue, net income, EBITDA 5. BALANCE_SHEET — assets, debt, equity, cash 6. CASH_FLOW — FCF, operating cash flow, capex 7. EARNINGS — EPS history, surprises 8. INSIDER_TRANSACTIONS — smart money buy/sell 9. REALTIME_OPTIONS — put/call ratio, IV, options sentiment 10. NEWS_SENTIMENT — news sentiment scores 11. ANALYTICS_FIXED_WINDOW — statistical return analytics 12. TREASURY_YIELD — 10Y yield (macro) 13. FEDERAL_FUNDS_RATE — interest rate environment 14. CPI — inflation 15. UNEMPLOYMENT — labor market 16. SPY daily — market context Prediction Methods: A. ML ensemble (RF + XGB + GBM + Elastic Net + MLP) — 21d return + quantile tails B. Technical momentum model — trend + mean-reversion signals C. Fundamental valuation model — fair value vs current price D. Sentiment composite — news + insider + options + earnings E. Macro overlay — applies sector/leverage multiplier to blended return F. Optional 3-model LLM debate — blended 45% into final expected return Final: Confidence-weighted blend of technical, fundamental, sentiment, and ML. Usage: python3 predict_monthly.py AAPL python3 predict_monthly.py MSFT --months 2 python3 predict_monthly.py AAPL --llm """ from llm_compat import claude_create # SDK-drift shim for Claude calls import argparse import json import os from collections import Counter os.environ.setdefault("PYTHONWARNINGS", "ignore") # suppress joblib/sklearn thread warnings import warnings warnings.filterwarnings("ignore") import csv import io import re import sys import time from datetime import datetime, timedelta from pathlib import Path import numpy as np import pandas as pd import requests from sklearn.decomposition import PCA from sklearn.inspection import permutation_importance from sklearn.ensemble import ( GradientBoostingRegressor, RandomForestRegressor, ) from sklearn.linear_model import ElasticNet, RidgeCV from sklearn.neural_network import MLPRegressor from sklearn.preprocessing import StandardScaler from scipy.stats import spearmanr import sklearn_parallel_hotfix # noqa: F401 from xgboost import XGBRegressor try: from predict_weekly import ( check_earnings_proximity, compute_post_earnings_drift_score, _raw_gpt_judge, _raw_kimi_judge, _raw_claude_judge, _parse_judge_response_weekly, _JUDGE_USER_DEBATE, _JUDGE_SYSTEM_DEBATE, ) except ImportError: def check_earnings_proximity(data): return None, "unknown", "" def compute_post_earnings_drift_score(*_a, **_k): return 0.0, "NEUTRAL", {} def _raw_gpt_judge(*_a, **_k): return None, 0, 0 def _raw_kimi_judge(*_a, **_k): return None, 0, 0 def _raw_claude_judge(*_a, **_k): return None, 0, 0 def _parse_judge_response_weekly(*_a, **_k): return None _JUDGE_USER_DEBATE = "{ticker}" _JUDGE_SYSTEM_DEBATE = "" # ─── Config ─────────────────────────────────────────────────────────────────── # Alpha Vantage key: environment only — never hardcode (the old literal key was # committed to git and must be treated as compromised/revoked). Set # ALPHAVANTAGE_API_KEY (or ALPHA_VANTAGE_API_KEY) in the shell or in .env. # NOTE: .env is loaded further down this module (load_dotenv), so the value is # re-read after that block; api_call() raises a clear error if it is still empty. API_KEY = (os.environ.get("ALPHAVANTAGE_API_KEY", "") or os.environ.get("ALPHA_VANTAGE_API_KEY", "")).strip() BASE_URL = "https://www.alphavantage.co/query" CACHE_DIR = Path(__file__).resolve().parent / "cache" CACHE_DIR.mkdir(exist_ok=True) # ─── FOMC / CPI (shared schedule; monthly uses wider detection window) ─────── _FOMC_ANNOUNCEMENT_DATES = [ "2025-01-29", "2025-03-19", "2025-05-07", "2025-06-18", "2025-07-30", "2025-09-17", "2025-10-29", "2025-12-10", "2026-01-28", "2026-03-18", "2026-05-06", "2026-06-17", "2026-07-29", "2026-09-16", "2026-10-28", "2026-12-09", ] _CPI_RELEASE_DATES = [ "2025-01-15", "2025-02-12", "2025-03-12", "2025-04-10", "2025-05-13", "2025-06-11", "2025-07-11", "2025-08-12", "2025-09-10", "2025-10-14", "2025-11-12", "2025-12-10", "2026-01-14", "2026-02-11", "2026-03-11", "2026-04-10", "2026-05-12", "2026-06-10", "2026-07-14", "2026-08-12", "2026-09-10", "2026-10-13", "2026-11-12", "2026-12-10", ] _SECTOR_MAP = { "Technology": 0, "Healthcare": 1, "Financials": 2, "Financial Services": 2, "Consumer Cyclical": 3, "Consumer Defensive": 4, "Industrials": 5, "Energy": 6, "Utilities": 7, "Real Estate": 8, "Basic Materials": 9, "Communication Services": 10, "": 11, } def _moonshot_assistant_text(message) -> str: for attr in ("content", "reasoning_content", "reasoning"): v = getattr(message, attr, None) if isinstance(v, str) and v.strip(): return v.strip() return "" def _get_llm_calibration_monthly(ticker, daily_df=None): cal_file = CACHE_DIR / f"{ticker}_llm_cal_monthly.json" if not cal_file.exists() or daily_df is None: return 1.0 try: history = json.loads(cal_file.read_text()) if not history: return 1.0 cutoff = datetime.now() - timedelta(days=60) correct = total = 0 for rec in history: pred_date = datetime.fromisoformat(rec["date"]) if pred_date < cutoff: continue pred_ts = pd.Timestamp(pred_date.date()) future = daily_df.index[daily_df.index >= pred_ts] if len(future) < 22: continue entry_ts = future[0] exit_ts = future[21] actual_ret = float(daily_df.loc[exit_ts, "adj_close"]) / float( daily_df.loc[entry_ts, "adj_close"] ) - 1 pred_dir = rec.get("direction", "NEUTRAL") if pred_dir == "BULLISH": correct += actual_ret > 0 total += 1 elif pred_dir == "BEARISH": correct += actual_ret < 0 total += 1 if total < 3: return 1.0 accuracy = correct / total return float(np.clip((accuracy - 0.5) * 2.0 + 1.0, 0.5, 1.2)) except Exception: return 1.0 _call_count = 0 _call_window = time.time() _last_call_ts = 0.0 # AlphaVantage rejects >5 requests/second with a "Burst pattern detected" Information payload. # The per-minute cap below doesn't prevent sub-second bursts, so enforce a per-call floor too. _MIN_CALL_INTERVAL = 0.25 # ~4 req/s — safely under the 5/s burst limit import threading as _threading _rate_lock = _threading.Lock() # serialize the throttle so concurrent fetchers still space calls BOLD = "\033[1m" DIM = "\033[2m" GREEN = "\033[92m" RED = "\033[91m" YELLOW = "\033[93m" CYAN = "\033[96m" MAGENTA = "\033[95m" BLUE = "\033[94m" WHITE = "\033[97m" RESET = "\033[0m" BG_GREEN = "\033[42m" BG_RED = "\033[41m" BG_YELLOW = "\033[43m" def _rate_limit(): global _call_count, _call_window, _last_call_ts # Lock so concurrent fetchers (prefetch pool) still space their dispatches and # share one honest per-minute counter instead of racing past AV's burst limit. with _rate_lock: # Per-second floor: space calls so a tight loop can't trip AV's 5-req/s burst limit. _since_last = time.time() - _last_call_ts if _since_last < _MIN_CALL_INTERVAL: time.sleep(_MIN_CALL_INTERVAL - _since_last) _last_call_ts = time.time() _call_count += 1 elapsed = time.time() - _call_window if elapsed < 60 and _call_count >= 73: # premium: 75/min, leave 2 margin wait = 62 - elapsed print(f" {DIM}(rate limit pause {wait:.0f}s){RESET}", end="", flush=True) time.sleep(wait) _call_count = 0 _call_window = time.time() elif elapsed >= 60: _call_count = 1 _call_window = time.time() def safe_float(v, default=0.0): """Convert to float, handling 'None' strings and other garbage.""" try: return float(v) except (ValueError, TypeError): return default def _sigmoid_prob(quantile_value, threshold, scale=25.0, cap=0.80): """Smooth mapping from quantile prediction to tail probability.""" x = (quantile_value - threshold) * scale return float(np.clip(cap / (1.0 + np.exp(-x)), 0.0, cap)) def _buyback_rate_yoy_from_shares(rows): """Negative = shares shrinking (buybacks). Expect list of {date, outstandingShares} or similar.""" if not rows or len(rows) < 2: return 0.0 try: parsed = [] for r in rows: d = r.get("date") or r.get("reportedDate", "") sh = safe_float(r.get("sharesOutstanding", r.get("outstandingShares", r.get("shares", 0)))) if d and sh > 0: parsed.append((pd.Timestamp(d), sh)) if len(parsed) < 2: return 0.0 parsed.sort(key=lambda x: x[0]) oldest, newest = parsed[0][1], parsed[-1][1] if oldest <= 0: return 0.0 return float((newest - oldest) / oldest) except Exception: return 0.0 def _iv_percentile_60d(hist_opts): """Rank current IV vs recent history if payload exposes implied volatility series.""" if not hist_opts: return 0.5 ivs = [] cur = None for r in hist_opts if isinstance(hist_opts, list) else []: iv = safe_float(r.get("impliedVolatility", r.get("implied_volatility", r.get("iv", 0)))) if iv > 0: ivs.append(iv) cur = iv if len(ivs) < 5 or cur is None: return 0.5 arr = np.array(ivs[-60:]) return float(np.mean(arr <= cur)) def _is_throttle_or_error_json(data): """True if an AlphaVantage JSON response is a throttle / error / empty payload that must NOT be cached or treated as data. AV signals rate limits via 'Information' (burst) and 'Note' (daily cap) — neither was previously caught, so error payloads poisoned the cache for the full TTL (e.g. 7-day OVERVIEW caches filled with 'Burst pattern detected').""" if not isinstance(data, dict): return True if "Information" in data or "Note" in data: return True if str(data.get("Error Message") or "").strip(): return True # Empty or metadata-only payloads (e.g. {"_ts": ...}) carry no usable data. if not any(k for k in data.keys() if not str(k).startswith("_")): return True return False def _is_throttle_or_error_csv(text): """CSV-mode equivalent: AV returns the throttle/error as a JSON-ish blurb instead of rows.""" head = (text or "")[:300] return (not head.strip()) or any( marker in head for marker in ("Error Message", "Information", "Note", "Thank you for using Alpha Vantage") ) def api_call(params, cache_key=None, cache_hours=4, no_cache=False, csv_response=False): """If csv_response=True, request CSV (datatype=csv) and return raw text (or cached string).""" # NB: throttle moved below the cache check — a cache hit hits no network, so it # must not pay the per-call floor (that was ~5s of dead time on a warm run). if csv_response: params = dict(params) params.setdefault("datatype", "csv") if cache_key and not no_cache: cf = CACHE_DIR / f"{cache_key}.json" if cf.exists(): try: d = json.loads(cf.read_text()) if (datetime.now() - datetime.fromisoformat(d["_ts"])).total_seconds() < cache_hours * 3600: d2 = d.copy() d2.pop("_ts", None) if csv_response and "_csv" in d2: if not _is_throttle_or_error_csv(d2["_csv"]): return d2["_csv"] elif not _is_throttle_or_error_json(d2): # Ignore a previously-poisoned cache entry and refetch. return d2 except Exception: pass if not API_KEY: # Raised only when a live request is actually needed (cache hits above still work). raise RuntimeError( "Alpha Vantage API key is not configured. Set ALPHAVANTAGE_API_KEY " "(or ALPHA_VANTAGE_API_KEY) in the environment or in the .env file next " "to this script. The hardcoded key fallback was removed for security." ) params["apikey"] = API_KEY _rate_limit() # only real network requests are throttled (cache hits returned above) try: r = requests.get(BASE_URL, params=params, timeout=30) if csv_response: text = r.text or "" if cache_key and not no_cache and not _is_throttle_or_error_csv(text): (CACHE_DIR / f"{cache_key}.json").write_text( json.dumps({"_csv": text, "_ts": datetime.now().isoformat()}, default=str) ) return text data = r.json() # Only cache genuine data — never AV throttle/error/empty payloads (prevents cache poisoning). if cache_key and not no_cache and not _is_throttle_or_error_json(data): to_save = data.copy(); to_save["_ts"] = datetime.now().isoformat() (CACHE_DIR / f"{cache_key}.json").write_text(json.dumps(to_save, default=str)) return data except Exception as e: return {"_error": str(e)} if not csv_response else "" def fetch_intraday_volume_profile(ticker, no_cache=False): """15-min bars → volume profile features (entitlement=delayed).""" raw = api_call( { "function": "TIME_SERIES_INTRADAY", "symbol": ticker, "interval": "15min", "outputsize": "full", "entitlement": "delayed", }, f"{ticker}_intraday_15min_monthly", 1, no_cache=no_cache, ) ts = raw.get("Time Series (15min)", {}) if not ts or len(ts) < 20: return None prices = [] volumes = [] for d, v in ts.items(): typical = (float(v["2. high"]) + float(v["3. low"]) + float(v["4. close"])) / 3 vol = float(v["5. volume"]) prices.append(typical) volumes.append(vol) if not prices: return None prices = np.array(prices) volumes = np.array(volumes) current_price = prices[0] n_bins = 30 price_min, price_max = prices.min(), prices.max() if price_max - price_min < 0.01: return None bin_edges = np.linspace(price_min, price_max, n_bins + 1) bin_volumes = np.zeros(n_bins) bin_centers = (bin_edges[:-1] + bin_edges[1:]) / 2 for p, v in zip(prices, volumes): idx = min(int((p - price_min) / (price_max - price_min) * n_bins), n_bins - 1) bin_volumes[idx] += v poc_idx = int(np.argmax(bin_volumes)) poc_price = float(bin_centers[poc_idx]) poc_distance = (current_price - poc_price) / (current_price + 1e-12) total_vol = bin_volumes.sum() target_vol = total_vol * 0.70 lo, hi = poc_idx, poc_idx area_vol = bin_volumes[poc_idx] while area_vol < target_vol and (lo > 0 or hi < n_bins - 1): expand_lo = bin_volumes[lo - 1] if lo > 0 else 0 expand_hi = bin_volumes[hi + 1] if hi < n_bins - 1 else 0 if expand_lo >= expand_hi and lo > 0: lo -= 1 area_vol += bin_volumes[lo] elif hi < n_bins - 1: hi += 1 area_vol += bin_volumes[hi] else: if lo > 0: lo -= 1 area_vol += bin_volumes[lo] else: break va_low = float(bin_edges[lo]) va_high = float(bin_edges[hi + 1]) if current_price > va_high: va_position = 1.0 elif current_price < va_low: va_position = -1.0 else: va_position = 0.0 avg_bin_vol = total_vol / n_bins hvn_mask = bin_volumes > 1.5 * avg_bin_vol hvn_centers = bin_centers[hvn_mask] hvn_support = 0.0 if len(hvn_centers) > 0: below = hvn_centers[hvn_centers < current_price] above = hvn_centers[hvn_centers > current_price] nearest_below = float(below.max()) if len(below) > 0 else None nearest_above = float(above.min()) if len(above) > 0 else None if nearest_below is not None and nearest_above is not None: dist_below = current_price - nearest_below dist_above = nearest_above - current_price hvn_support = 1.0 if dist_below < dist_above else -1.0 elif nearest_below is not None: hvn_support = 1.0 elif nearest_above is not None: hvn_support = -1.0 return { "poc_distance": float(poc_distance), "value_area_position": float(va_position), "hvn_support": float(hvn_support), } # ─── Data Collection (all 16 endpoints) ────────────────────────────────────── # ─── Analysis Modules ───────────────────────────────────────────────────────── def _build_qualitative_assessments(signals, de_thresh=None): """Build qualitative Good/Neutral/Bad assessment for each fundamental metric. Returns list of (category, metric_name, value_str, assessment, note) tuples. Assessment is one of: STRONG, GOOD, NEUTRAL, WEAK, BAD, N/A. """ if de_thresh is None: de_thresh = _SECTOR_DE_THRESHOLDS["DEFAULT"] assessments = [] def _add(cat, name, key, fmt, thresholds, invert=False): """Helper: thresholds = [(val, "STRONG"), (val, "GOOD"), ...] descending. If invert=True, lower is better (e.g., P/E, D/E).""" val = signals.get(key) if val is None: assessments.append((cat, name, "N/A", "N/A", "")) return val_str = fmt(val) assessment = "NEUTRAL" note = "" for threshold, label in thresholds: if invert: if val <= threshold: assessment = label break else: if val >= threshold: assessment = label break else: # Fell through — worst case for both inverted and non-inverted metrics # Inverted (lower is better): value exceeded all thresholds → too expensive → worst label # Non-inverted (higher is better): value below all thresholds → too low → worst label assessment = thresholds[-1][1] assessments.append((cat, name, val_str, assessment, note)) pct = lambda v: f"{v*100:.1f}%" dec1 = lambda v: f"{v:.1f}" dec2 = lambda v: f"{v:.2f}" money = lambda v: f"${v/1e9:.1f}B" if abs(v) >= 1e9 else f"${v/1e6:.0f}M" # ── Valuation ── _add("Valuation", "P/E Ratio", "pe_ratio", dec1, [(15, "STRONG"), (20, "GOOD"), (25, "NEUTRAL"), (35, "WEAK")], invert=True) _add("Valuation", "PEG Ratio", "peg_ratio", dec2, [(1.0, "STRONG"), (1.5, "GOOD"), (2.0, "NEUTRAL"), (3.0, "WEAK")], invert=True) _add("Valuation", "P/FCF", "p_fcf", dec1, [(15, "STRONG"), (20, "GOOD"), (30, "NEUTRAL"), (50, "WEAK")], invert=True) _add("Valuation", "P/S Ratio", "price_to_sales", dec1, [(2, "STRONG"), (5, "GOOD"), (10, "NEUTRAL"), (20, "WEAK")], invert=True) _add("Valuation", "P/B Ratio", "price_to_book", dec1, [(1.5, "STRONG"), (3, "GOOD"), (5, "NEUTRAL"), (10, "WEAK")], invert=True) _add("Valuation", "EV/EBITDA", "ev_to_ebitda", dec1, [(10, "STRONG"), (15, "GOOD"), (20, "NEUTRAL"), (30, "WEAK")], invert=True) _add("Valuation", "EV/Revenue", "ev_to_revenue", dec1, [(2, "STRONG"), (5, "GOOD"), (10, "NEUTRAL"), (20, "WEAK")], invert=True) _add("Valuation", "52W High Drop", "52w_high_drop", pct, [(0.50, "STRONG"), (0.25, "GOOD"), (0.10, "NEUTRAL")], invert=False) _add("Valuation", "Analyst Upside", "analyst_upside", pct, [(0.30, "STRONG"), (0.15, "GOOD"), (0.05, "NEUTRAL"), (-0.05, "WEAK")], invert=False) # ── Growth ── _add("Growth", "Revenue Growth YoY", "revenue_growth_yoy", pct, [(0.20, "STRONG"), (0.10, "GOOD"), (0.05, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Growth", "Earnings Growth YoY", "earnings_growth_yoy", pct, [(0.20, "STRONG"), (0.10, "GOOD"), (0.05, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Growth", "FCF Growth YoY", "fcf_growth_yoy", pct, [(0.20, "STRONG"), (0.10, "GOOD"), (0, "NEUTRAL"), (-0.20, "WEAK")], invert=False) # ── Profitability ── _add("Profitability", "ROIC", "roic", pct, [(0.20, "STRONG"), (0.15, "GOOD"), (0.10, "NEUTRAL"), (0.05, "WEAK")], invert=False) _add("Profitability", "ROE", "roe", pct, [(0.20, "STRONG"), (0.15, "GOOD"), (0.10, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Profitability", "Operating Margin", "operating_margin", pct, [(0.25, "STRONG"), (0.15, "GOOD"), (0.10, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Profitability", "Profit Margin", "profit_margin", pct, [(0.20, "STRONG"), (0.10, "GOOD"), (0.05, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Profitability", "Gross Margin", "gross_margin", pct, [(0.60, "STRONG"), (0.40, "GOOD"), (0.25, "NEUTRAL"), (0.10, "WEAK")], invert=False) # ── Financial Health ── # D/E thresholds are sector-dependent — financials/utilities carry structural leverage _add("Health", "Debt/Equity", "debt_to_equity", dec2, [(de_thresh["strong"], "STRONG"), (de_thresh["good"], "GOOD"), (de_thresh["neutral"], "NEUTRAL"), (de_thresh["neutral"] * 2, "WEAK")], invert=True) _add("Health", "Interest Coverage", "interest_coverage", lambda v: "Debt-free" if v == float('inf') else ("N/A" if v is None else f"{v:.1f}x"), [(10, "STRONG"), (5, "GOOD"), (3, "NEUTRAL"), (1.5, "WEAK")], invert=False) _add("Health", "Current Ratio", "current_ratio", dec2, [(2.0, "STRONG"), (1.5, "GOOD"), (1.0, "NEUTRAL"), (0.8, "WEAK")], invert=False) # ── Cash Flow ── _add("Cash Flow", "FCF Margin", "fcf_margin", pct, [(0.15, "STRONG"), (0.10, "GOOD"), (0.05, "NEUTRAL"), (0, "WEAK")], invert=False) _add("Cash Flow", "FCF Conversion", "fcf_conversion", pct, [(1.0, "STRONG"), (0.80, "GOOD"), (0.50, "NEUTRAL"), (0, "WEAK")], invert=False) ocf_gt_ni = signals.get("ocf_gt_ni") if ocf_gt_ni is not None: assessments.append(("Cash Flow", "OCF > Net Income", "Yes" if ocf_gt_ni else "No", "GOOD" if ocf_gt_ni else "BAD", "earnings quality")) # ── Dividends ── div_y = signals.get("dividend_yield", 0) payout = signals.get("payout_ratio", 0) if div_y > 0: _add("Dividends", "Dividend Yield", "dividend_yield", pct, [(0.04, "STRONG"), (0.02, "GOOD"), (0.01, "NEUTRAL")], invert=False) _add("Dividends", "Payout Ratio", "payout_ratio", pct, [(0.75, "BAD"), (0.60, "NEUTRAL"), (0.40, "GOOD"), (0, "STRONG")], invert=True) else: assessments.append(("Dividends", "Dividend", "None", "NEUTRAL", "no dividend")) # ── Risk ── _add("Risk", "Beta", "beta", dec2, [(1.5, "BAD"), (1.2, "WEAK"), (0.8, "NEUTRAL"), (0.5, "GOOD")], invert=True) # ── Earnings Quality ── br = signals.get("beat_rate") if br is not None: label = "STRONG" if br >= 0.75 else "GOOD" if br >= 0.50 else "WEAK" if br >= 0.25 else "BAD" assessments.append(("Earnings", "Beat Rate (last 4 qtrs)", f"{br*100:.0f}%", label, "")) surprise = signals.get("avg_earnings_surprise") if surprise is not None: label = "STRONG" if surprise > 5 else "GOOD" if surprise > 0 else "WEAK" if surprise > -5 else "BAD" assessments.append(("Earnings", "Avg Surprise", f"{surprise:+.1f}%", label, "")) return assessments # D/E thresholds by sector — financial companies operate with structural leverage _SECTOR_DE_THRESHOLDS = { "Financials": {"strong": 6.0, "good": 10.0, "neutral": 15.0}, "Financial Services": {"strong": 5.0, "good": 9.0, "neutral": 14.0}, "Real Estate": {"strong": 1.5, "good": 2.5, "neutral": 4.0}, "Utilities": {"strong": 1.2, "good": 2.0, "neutral": 3.0}, "DEFAULT": {"strong": 0.3, "good": 0.5, "neutral": 1.0}, } def analyze_fundamentals(data): """Comprehensive quant scorecard — 6 categories, 32 max points.""" ov = data.get("overview", {}) if not ov: return None signals = {} scorecard = {} # category → {metric: points} MAX_POINTS = 32.0 sector = (ov.get("Sector") or ov.get("GicsSector") or "").strip() de_thresh = _SECTOR_DE_THRESHOLDS.get(sector, _SECTOR_DE_THRESHOLDS["DEFAULT"]) # ── helper: extract TTM from quarterly reports ── def _ttm(reports, field, n=4): """Sum last n quarterly values for trailing-twelve-months.""" total = 0.0 count = 0 for r in (reports or [])[:n]: v = safe_float(r.get(field, 0)) if v != 0: total += v count += 1 return total if count >= max(n - 1, 1) else None def _latest(reports, field): """Most recent quarterly value.""" if not reports: return None return safe_float(reports[0].get(field, 0)) or None def _prev(reports, field, offset=4): """Same-quarter prior year value for YoY comparison.""" if not reports or len(reports) <= offset: return None return safe_float(reports[offset].get(field, 0)) or None # Raw data sources inc_q = data.get("income", {}).get("quarterlyReports", []) bs_q = data.get("balance", {}).get("quarterlyReports", []) cf_q = data.get("cashflow", {}).get("quarterlyReports", []) # Get current price from daily data or overview daily_df_price = data.get("daily", None) if daily_df_price is not None and not daily_df_price.empty: close_col_p = "adj_close" if "adj_close" in daily_df_price.columns else \ "adjusted_close" if "adjusted_close" in daily_df_price.columns else daily_df_price.columns[0] current_price = float(daily_df_price[close_col_p].iloc[-1]) else: current_price = safe_float(ov.get("AnalystTargetPrice", 0)) # rough fallback # ══════════════════════════════════════════════════════════════ # 1. VALUATION (Max 8 points) # ══════════════════════════════════════════════════════════════ val_pts = {} # 52-week high drop (from daily DataFrame) daily_df = data.get("daily", None) if daily_df is not None and not daily_df.empty and current_price > 0: close_col = "adj_close" if "adj_close" in daily_df.columns else \ "adjusted_close" if "adjusted_close" in daily_df.columns else daily_df.columns[0] recent_252 = daily_df[close_col].dropna().tail(252) if len(recent_252) > 50: high_52w = recent_252.max() if high_52w > 0: drop_pct = (high_52w - current_price) / high_52w signals["52w_high_drop"] = drop_pct if drop_pct > 0.50: val_pts["52w_drop"] = 2 elif drop_pct > 0.25: val_pts["52w_drop"] = 1 else: val_pts["52w_drop"] = 0 # P/E Ratio pe = safe_float(ov.get("PERatio", 0)) fwd_pe = safe_float(ov.get("ForwardPE", 0)) if pe > 0: signals["pe_ratio"] = pe if pe < 15: val_pts["pe"] = 2 elif pe <= 25: val_pts["pe"] = 1 else: val_pts["pe"] = 0 if fwd_pe > 0 and pe > 0: signals["pe_fwd_discount"] = (pe - fwd_pe) / pe # PEG Ratio peg = safe_float(ov.get("PEGRatio", 0)) if peg > 0: signals["peg_ratio"] = peg if peg < 1.0: val_pts["peg"] = 2 elif peg <= 1.5: val_pts["peg"] = 1 else: val_pts["peg"] = 0 # Price to Free Cash Flow (P/FCF) ttm_fcf = None ttm_ocf = _ttm(cf_q, "operatingCashflow") ttm_capex = _ttm(cf_q, "capitalExpenditures") if ttm_ocf is not None and ttm_capex is not None: ttm_fcf = ttm_ocf - abs(ttm_capex) # capex often reported negative signals["ttm_fcf"] = ttm_fcf shares = safe_float(ov.get("SharesOutstanding", 0)) if ttm_fcf > 0 and current_price > 0 and shares > 0: p_fcf = (current_price * shares) / ttm_fcf signals["p_fcf"] = p_fcf if p_fcf < 20: val_pts["p_fcf"] = 2 elif p_fcf <= 30: val_pts["p_fcf"] = 1 else: val_pts["p_fcf"] = 0 scorecard["Valuation"] = (sum(val_pts.values()), 8, val_pts) # ══════════════════════════════════════════════════════════════ # 2. GROWTH (Max 6 points) # ══════════════════════════════════════════════════════════════ gro_pts = {} # Revenue Growth YoY rev_growth = safe_float(ov.get("QuarterlyRevenueGrowthYOY", 0)) if rev_growth != 0: signals["revenue_growth_yoy"] = rev_growth if rev_growth > 0.20: gro_pts["rev_growth"] = 2 elif rev_growth > 0.10: gro_pts["rev_growth"] = 1 else: gro_pts["rev_growth"] = 0 # Earnings Growth YoY earn_growth = safe_float(ov.get("QuarterlyEarningsGrowthYOY", 0)) if earn_growth != 0: signals["earnings_growth_yoy"] = earn_growth if earn_growth > 0.20: gro_pts["earn_growth"] = 2 elif earn_growth > 0.10: gro_pts["earn_growth"] = 1 else: gro_pts["earn_growth"] = 0 # Free Cash Flow Growth YoY if ttm_fcf is not None: prev_ocf = _ttm(cf_q[4:], "operatingCashflow") if len(cf_q) > 4 else None prev_capex = _ttm(cf_q[4:], "capitalExpenditures") if len(cf_q) > 4 else None if prev_ocf is not None and prev_capex is not None: prev_fcf = prev_ocf - abs(prev_capex) if prev_fcf > 0: fcf_growth = (ttm_fcf - prev_fcf) / abs(prev_fcf) signals["fcf_growth_yoy"] = fcf_growth if fcf_growth > 0.20: gro_pts["fcf_growth"] = 2 elif fcf_growth > 0.10: gro_pts["fcf_growth"] = 1 else: gro_pts["fcf_growth"] = 0 scorecard["Growth"] = (sum(gro_pts.values()), 6, gro_pts) # ══════════════════════════════════════════════════════════════ # 3. PROFITABILITY (Max 8 points) # ══════════════════════════════════════════════════════════════ prof_pts = {} # ROIC = NOPAT / Invested Capital ttm_ebit = _ttm(inc_q, "ebit") tax_rate = 0.21 # assume statutory latest_total_debt = _latest(bs_q, "shortLongTermDebtTotal") or ( (_latest(bs_q, "shortTermDebt") or 0) + (_latest(bs_q, "longTermDebt") or 0)) latest_equity = _latest(bs_q, "totalShareholderEquity") if ttm_ebit and latest_equity: nopat = ttm_ebit * (1 - tax_rate) invested_capital = (latest_equity or 0) + (latest_total_debt or 0) if invested_capital > 0: roic = nopat / invested_capital signals["roic"] = roic if roic > 0.15: prof_pts["roic"] = 2 elif roic > 0.10: prof_pts["roic"] = 1 else: prof_pts["roic"] = 0 # ROE roe = safe_float(ov.get("ReturnOnEquityTTM", 0)) if roe != 0: signals["roe"] = roe if roe > 0.20: prof_pts["roe"] = 2 elif roe > 0.15: prof_pts["roe"] = 1 else: prof_pts["roe"] = 0 # Operating Margin op_margin = safe_float(ov.get("OperatingMarginTTM", 0)) profit_margin = safe_float(ov.get("ProfitMargin", 0)) signals["profit_margin"] = profit_margin if op_margin != 0: signals["operating_margin"] = op_margin if op_margin > 0.20: prof_pts["op_margin"] = 2 elif op_margin > 0.15: prof_pts["op_margin"] = 1 else: prof_pts["op_margin"] = 0 elif profit_margin != 0: # fallback to profit margin if operating margin unavailable if profit_margin > 0.20: prof_pts["op_margin"] = 2 elif profit_margin > 0.15: prof_pts["op_margin"] = 1 else: prof_pts["op_margin"] = 0 # Gross Margin (bonus context, not scored but tracked) gross_margin = safe_float(ov.get("GrossProfitTTM", 0)) ttm_rev = _ttm(inc_q, "totalRevenue") if gross_margin > 0 and ttm_rev and ttm_rev > 0: signals["gross_margin"] = gross_margin / ttm_rev scorecard["Profitability"] = (sum(prof_pts.values()), 8, prof_pts) # ══════════════════════════════════════════════════════════════ # 4. FINANCIAL HEALTH (Max 4 points) # ══════════════════════════════════════════════════════════════ health_pts = {} # Debt to Equity — thresholds depend on sector (financials have structural leverage) if latest_equity and latest_equity > 0 and latest_total_debt is not None: d_e = latest_total_debt / latest_equity signals["debt_to_equity"] = d_e if d_e < de_thresh["strong"]: health_pts["d_e"] = 2 elif d_e <= de_thresh["good"]: health_pts["d_e"] = 1 else: health_pts["d_e"] = 0 # Interest Coverage = EBIT / Interest Expense # Try multiple field names — AV uses different names by sector ttm_interest = (_ttm(inc_q, "interestExpense") or _ttm(inc_q, "interestAndDebtExpense") or _ttm(inc_q, "totalInterestExpense")) # Sanity check: if company has meaningful debt, can't be "Debt-free" has_debt = (signals.get("debt_to_equity", 0) or 0) > 0.1 if ttm_ebit and ttm_ebit > 0: if ttm_interest and ttm_interest > 0: int_coverage = ttm_ebit / ttm_interest signals["interest_coverage"] = int_coverage if int_coverage > 8: health_pts["int_cov"] = 2 elif int_coverage >= 5: health_pts["int_cov"] = 1 else: health_pts["int_cov"] = 0 elif not has_debt: # Truly debt-free: no interest expense AND no meaningful debt signals["interest_coverage"] = float('inf') health_pts["int_cov"] = 2 else: # Has debt but interest expense not found in income statement # (common for financial companies that report it differently) # Score 0 — cannot verify coverage, treat as unknown/risky signals["interest_coverage"] = None health_pts["int_cov"] = 0 scorecard["Financial Health"] = (sum(health_pts.values()), 4, health_pts) # ══════════════════════════════════════════════════════════════ # 5. DIVIDENDS (Max 2 points) # ══════════════════════════════════════════════════════════════ div_pts = {} div_yield = safe_float(ov.get("DividendYield", 0)) signals["dividend_yield"] = div_yield payout_ratio = safe_float(ov.get("PayoutRatio")) or 0.0 # Fall back to DividendPerShare / EPS when Overview API returns None if payout_ratio == 0.0: _dps = safe_float(ov.get("DividendPerShare")) or 0.0 _eps = safe_float(ov.get("EPS")) or 0.0 if _dps > 0 and _eps > 0: payout_ratio = _dps / _eps signals["payout_ratio"] = payout_ratio if payout_ratio > 0 and payout_ratio <= 0.75: div_pts["payout"] = 2 elif payout_ratio == 0 and div_yield == 0: div_pts["payout"] = 0 # no dividend, neutral else: div_pts["payout"] = 0 # payout > 75% or negative scorecard["Dividends"] = (sum(div_pts.values()), 2, div_pts) # ══════════════════════════════════════════════════════════════ # 6. CASH FLOW QUALITY (Max 4 points) # ══════════════════════════════════════════════════════════════ cf_pts = {} # Positive FCF if ttm_fcf is not None: if ttm_fcf > 0: cf_pts["positive_fcf"] = 1 else: cf_pts["positive_fcf"] = 0 # Earnings Quality: OCF > Net Income ttm_net_income = _ttm(inc_q, "netIncome") if ttm_ocf is not None and ttm_net_income is not None: signals["ttm_net_income"] = ttm_net_income if ttm_ocf > ttm_net_income: cf_pts["earnings_quality"] = 1 signals["ocf_gt_ni"] = True else: cf_pts["earnings_quality"] = 0 signals["ocf_gt_ni"] = False # FCF Margin = FCF / Revenue if ttm_fcf is not None and ttm_rev and ttm_rev > 0: fcf_margin = ttm_fcf / ttm_rev signals["fcf_margin"] = fcf_margin if fcf_margin > 0.10: cf_pts["fcf_margin"] = 1 else: # Check if improving YoY prev_fcf_for_margin = None prev_rev = _ttm(inc_q[4:], "totalRevenue") if len(inc_q) > 4 else None prev_ocf_m = _ttm(cf_q[4:], "operatingCashflow") if len(cf_q) > 4 else None prev_capex_m = _ttm(cf_q[4:], "capitalExpenditures") if len(cf_q) > 4 else None if prev_ocf_m is not None and prev_capex_m is not None and prev_rev and prev_rev > 0: prev_fcf_for_margin = prev_ocf_m - abs(prev_capex_m) prev_fcf_margin = prev_fcf_for_margin / prev_rev if fcf_margin > prev_fcf_margin: cf_pts["fcf_margin"] = 0.5 signals["fcf_margin_improving"] = True else: cf_pts["fcf_margin"] = 0 else: cf_pts["fcf_margin"] = 0 # FCF Conversion = FCF / Net Income if ttm_fcf is not None and ttm_net_income and ttm_net_income > 0: fcf_conversion = ttm_fcf / ttm_net_income signals["fcf_conversion"] = fcf_conversion if fcf_conversion > 0.80: cf_pts["fcf_conversion"] = 1 else: # Check if improving YoY if "fcf_growth_yoy" in signals: # we have prior year FCF prev_ocf_c = _ttm(cf_q[4:], "operatingCashflow") if len(cf_q) > 4 else None prev_capex_c = _ttm(cf_q[4:], "capitalExpenditures") if len(cf_q) > 4 else None prev_ni = _ttm(inc_q[4:], "netIncome") if len(inc_q) > 4 else None if prev_ocf_c is not None and prev_capex_c is not None and prev_ni and prev_ni > 0: prev_conv = (prev_ocf_c - abs(prev_capex_c)) / prev_ni if fcf_conversion > prev_conv: cf_pts["fcf_conversion"] = 0.5 signals["fcf_conversion_improving"] = True else: cf_pts["fcf_conversion"] = 0 else: cf_pts["fcf_conversion"] = 0 else: cf_pts["fcf_conversion"] = 0 scorecard["Cash Flow Quality"] = (sum(cf_pts.values()), 4, cf_pts) # ══════════════════════════════════════════════════════════════ # BONUS: Analyst consensus + earnings surprises (context, not in 32pts) # ══════════════════════════════════════════════════════════════ analyst_bonus = 0.0 # Analyst target try: target = safe_float(ov.get("AnalystTargetPrice", 0)) shares = safe_float(ov.get("SharesOutstanding", 0)) if target > 0 and current_price > 0: upside = (target - current_price) / current_price signals["analyst_upside"] = upside analyst_bonus += np.clip(upside * 0.15, -0.08, 0.08) except (ValueError, TypeError, ZeroDivisionError): pass # Earnings surprises if "earnings" in data and "quarterlyEarnings" in data["earnings"]: qe = data["earnings"]["quarterlyEarnings"][:4] surprises = [] for q in qe: try: surprises.append(safe_float(q.get("surprisePercentage", 0))) except Exception: pass if surprises: avg_surprise = np.mean(surprises) beats = sum(1 for s in surprises if s > 0) signals["avg_earnings_surprise"] = avg_surprise signals["beat_rate"] = beats / len(surprises) analyst_bonus += np.clip(avg_surprise / 50, -0.05, 0.05) # ══════════════════════════════════════════════════════════════ # FINAL SCORE → expected return # ══════════════════════════════════════════════════════════════ total_pts = sum(cat[0] for cat in scorecard.values()) pct_score = total_pts / MAX_POINTS # 0.0 to 1.0 # Map score to expected return: # 0% → -8% (terrible company) # 50% → 0% (average) # 100% → +10% (exceptional) base_return = (pct_score - 0.5) * 0.18 # range: -9% to +9% # Scale analyst bonus by scorecard quality so a poor score can't be # fully overridden by analyst consensus quality_scale = 0.2 + 0.8 * pct_score analyst_bonus *= quality_scale expected_return = np.clip(base_return + analyst_bonus, -0.10, 0.10) # Confidence: based on data completeness (how many metrics we could calculate) total_metrics_possible = 14 # total sub-metrics across all categories total_metrics_scored = sum(len(cat[2]) for cat in scorecard.values()) data_coverage = min(total_metrics_scored / total_metrics_possible, 1.0) # ══════════════════════════════════════════════════════════════ # ADDITIONAL METRICS (not scored, but tracked for LLM context) # ══════════════════════════════════════════════════════════════ ps = safe_float(ov.get("PriceToSalesRatioTTM", 0)) if ps > 0: signals["price_to_sales"] = ps pb = safe_float(ov.get("PriceToBookRatio", 0)) if pb > 0: signals["price_to_book"] = pb ev_ebitda = safe_float(ov.get("EVToEBITDA", 0)) if ev_ebitda > 0: signals["ev_to_ebitda"] = ev_ebitda ev_rev = safe_float(ov.get("EVToRevenue", 0)) if ev_rev > 0: signals["ev_to_revenue"] = ev_rev beta = safe_float(ov.get("Beta", 0)) if beta > 0: signals["beta"] = beta eps = safe_float(ov.get("EPS", 0)) if eps != 0: signals["eps"] = eps rev_ttm = safe_float(ov.get("RevenueTTM", 0)) if rev_ttm > 0: signals["revenue_ttm"] = rev_ttm ebitda = safe_float(ov.get("EBITDA", 0)) if ebitda > 0: signals["ebitda"] = ebitda # Current Ratio from balance sheet current_assets = _latest(bs_q, "totalCurrentAssets") current_liabilities = _latest(bs_q, "totalCurrentLiabilities") if current_assets and current_liabilities and current_liabilities > 0: signals["current_ratio"] = current_assets / current_liabilities # ══════════════════════════════════════════════════════════════ # QUALITATIVE ASSESSMENT TABLE (for LLM consumption) # ══════════════════════════════════════════════════════════════ assessments = _build_qualitative_assessments(signals, de_thresh) return { "expected_return": expected_return, "score": pct_score, "total_points": total_pts, "max_points": MAX_POINTS, "scorecard": scorecard, "signals": signals, "assessments": assessments, "confidence": max(data_coverage, 0.20), # floor at 20% — sparse data should carry less weight } def _compute_enhanced_news_sentiment(data): """Enhanced news sentiment with recency weighting, relevance filtering, and momentum (trend) detection. Returns dict with keys: weighted_avg, raw_avg, momentum, n_articles, recent_avg (last 7d), older_avg (8-30d), or None if no data. """ news_items = data.get("news_items", []) if not news_items: # Fallback to legacy scores if news_items not available scores = data.get("news_scores", []) if not scores: return None return { "weighted_avg": float(np.mean(scores)), "raw_avg": float(np.mean(scores)), "momentum": 0.0, "n_articles": len(scores), "recent_avg": float(np.mean(scores)), "older_avg": float(np.mean(scores)), } now = datetime.now() scored = [] for item in news_items: sent = item["score"] rel = item.get("relevance", 0.5) tp = item.get("time_published", "") # Parse Alpha Vantage timestamp: "20260329T120000" days_ago = 15 # default if unparseable if tp: try: pub_dt = datetime.strptime(tp[:15], "%Y%m%dT%H%M%S") days_ago = max((now - pub_dt).total_seconds() / 86400, 0.1) except (ValueError, IndexError): pass # Exponential recency decay: half-life 7 days recency_weight = np.exp(-0.693 * days_ago / 7.0) # ln(2)/7 ≈ 0.099 # Relevance filter: articles below 0.2 relevance get down-weighted rel_weight = max(rel, 0.1) combined_weight = recency_weight * rel_weight scored.append((sent, combined_weight, days_ago)) if not scored: return None # Weighted average sentiment total_w = sum(w for _, w, _ in scored) weighted_avg = sum(s * w for s, w, _ in scored) / total_w if total_w > 0 else 0 # Raw average (for comparison) raw_avg = float(np.mean([s for s, _, _ in scored])) # Recent vs older split for momentum recent = [s for s, _, d in scored if d <= 7] older = [s for s, _, d in scored if d > 7] recent_avg = float(np.mean(recent)) if recent else weighted_avg older_avg = float(np.mean(older)) if older else weighted_avg # Momentum: positive = sentiment improving, negative = deteriorating momentum = recent_avg - older_avg return { "weighted_avg": weighted_avg, "raw_avg": raw_avg, "momentum": momentum, "n_articles": len(scored), "n_recent": len(recent), "n_older": len(older), "recent_avg": recent_avg, "older_avg": older_avg, } def _insider_score_from_rows(insider, lookback_days=90, return_counts=False): """Insider buy/sell pressure ∈ [-1, +1] from INSIDER_TRANSACTIONS rows over the last `lookback_days`. -1 = pure selling, +1 = pure buying. Returns None if no usable rows. Shared by analyze_sentiment (Stage 2) and the Stage-1.5 distribution pre-screen so both use an identical definition. Blends transaction-count ratio (0.4) and dollar-value ratio (0.6). return_counts=True returns (score, buys, sells) for callers that need the breakdown; otherwise returns the scalar score (or None / (None,0,0) when there are no usable rows). """ if not insider: return (None, 0, 0) if return_counts else None now = datetime.now() buys, sells, buy_val, sell_val = 0, 0, 0.0, 0.0 # Filter by the date window across ALL rows (not a fixed [:100] slice): high-volume names # (AAPL/IBM have thousands of filings) could otherwise have a 90-day window truncated mid- # window, biasing the score toward whatever the newest cluster happens to be. for txn in insider: try: td = datetime.strptime(txn.get("transaction_date", ""), "%Y-%m-%d") except ValueError: continue if (now - td).days > lookback_days: continue shares = safe_float(txn.get("shares", 0)) price = safe_float(txn.get("share_price", 0)) if txn.get("acquisition_or_disposal") == "A": buys += 1; buy_val += shares * price elif txn.get("acquisition_or_disposal") == "D": sells += 1; sell_val += shares * price total = buys + sells if total == 0: return (None, 0, 0) if return_counts else None ratio = buys / total val_ratio = buy_val / (buy_val + sell_val) if (buy_val + sell_val) > 0 else 0.5 score = (ratio * 0.4 + val_ratio * 0.6) * 2 - 1 return (score, buys, sells) if return_counts else score def analyze_sentiment(data, ticker): """Composite sentiment from news + insider + options + earnings. News component uses recency-weighted, relevance-filtered scores with momentum detection (recent vs older sentiment trend). """ components = {} score = 0.0 weight_sum = 0.0 # ── News sentiment (enhanced) ── news = _compute_enhanced_news_sentiment(data) if news: components["news_sentiment"] = news["weighted_avg"] components["news_raw_avg"] = news["raw_avg"] components["news_momentum"] = news["momentum"] components["news_n_articles"] = news["n_articles"] components["news_recent_avg"] = news["recent_avg"] components["news_older_avg"] = news["older_avg"] # Blend: 70% weighted sentiment + 30% momentum signal news_signal = news["weighted_avg"] * 0.7 + news["momentum"] * 0.3 score += news_signal * 0.3 weight_sum += 0.3 # ── Insider transactions (90 day) ── insider = data.get("insider", []) if insider: insider_score, buys, sells = _insider_score_from_rows(insider, return_counts=True) if insider_score is not None: components["insider_score"] = insider_score score += insider_score * 0.3 weight_sum += 0.3 components["insider_detail"] = f"buys:{buys} sells:{sells}" # ── Options put/call ── options = data.get("options", []) if options: call_vol = sum(safe_float(o.get("volume", 0)) for o in options if o.get("type", "").lower() == "call") put_vol = sum(safe_float(o.get("volume", 0)) for o in options if o.get("type", "").lower() == "put") if call_vol > 0: pc_ratio = put_vol / call_vol options_score = np.clip(-(pc_ratio - 1.0) * 2, -1, 1) components["options_pc_ratio"] = pc_ratio components["options_score"] = options_score score += options_score * 0.2 weight_sum += 0.2 # IV level ivs = [safe_float(o.get("implied_volatility", 0)) for o in options if safe_float(o.get("implied_volatility", 0)) > 0] if ivs: components["avg_iv"] = np.mean(ivs) # ── Earnings momentum ── if "earnings" in data and "quarterlyEarnings" in data["earnings"]: qe = data["earnings"]["quarterlyEarnings"][:4] surprises = [safe_float(q.get("surprisePercentage", 0)) for q in qe if q.get("surprisePercentage") not in (None, "None", "")] if surprises: earn_score = np.clip(np.mean(surprises) / 10, -1, 1) * 0.5 + (sum(1 for s in surprises if s > 0) / len(surprises) - 0.5) components["earnings_score"] = earn_score score += earn_score * 0.2 weight_sum += 0.2 if weight_sum == 0: return None norm_score = score / weight_sum expected_return = np.clip(norm_score * 0.06, -0.10, 0.10) # Dampen confidence when sentiment components disagree internally _component_vals = [] for _ck in ("news_sentiment", "insider_score", "options_score", "earnings_score"): _cv = components.get(_ck) if isinstance(_cv, (int, float)) and abs(_cv) > 0.05: _component_vals.append(float(_cv)) if len(_component_vals) >= 2: _pos = sum(1 for v in _component_vals if v > 0) _neg = sum(1 for v in _component_vals if v < 0) _agreement = max(_pos, _neg) / len(_component_vals) _disagreement_penalty = 0.5 + 0.5 * _agreement weight_sum *= _disagreement_penalty return { "expected_return": expected_return, "score": norm_score, "components": components, "confidence": min(weight_sum / 1.0, 1.0), } # ─── ML Monthly Return Predictor ───────────────────────────────────────────── # Filled by build_ml_features — used by train_ml_model for dropna(subset=core) MONTHLY_NULLABLE_BASES = set() _SCALED_MODELS = {"enet", "mlp"} # models requiring StandardScaler'd input # Stacked meta-learner: floor weights so no model is dropped; fall back if meta inverts sign vs base models. _META_MIN_WEIGHT = 0.05 # Stacked meta is fit on few OOS folds and can extrapolate far past the base-model consensus. # Constrain its output to within this band (absolute 21d return) of the dir-acc-gated weighted avg. _META_DIVERGENCE_TOL = 0.04 # Cap any single base learner's ensemble weight so one degenerate model cannot dominate. _MAX_MODEL_WEIGHT = 0.35 # Features excluded from the ML training matrix despite being economically motivated, because a # feature-dominance audit (per-ticker permutation importance vs forward-21d rank-IC across 12 # names) flagged them as overfit surfaces: meaningful model importance but near-zero, sign- # inconsistent IC across tickers (trees memorizing per-ticker noise, not generalizable signal). # • days_to_next_earnings — ~0.32 importance, IC −0.10..+0.02 (smooth 0–90 earnings-cycle # sawtooth). Earnings-proximity risk is already handled explicitly (alpha penalties + # risk_score + check_earnings_proximity); the coarse binary `earnings_in_next_21d` is kept. # • skew_63d — 5.1% importance but mean|IC| 0.033 with coin-flip sign (signCons 0.50): a # higher-moment return-shape feature with no stable cross-ticker direction. _ML_EXCLUDE_FEATURES = {"days_to_next_earnings", "skew_63d"} # ─── LLM Judge (GPT-5.5 / Gemini Flash) ───────────────────────────────────── # Load .env if available (keeps keys out of shell history) try: from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"), override=True) except ImportError: pass # Re-read the Alpha Vantage key now that .env has been loaded (the module-top read # at the Config section only sees the shell environment). API_KEY = (os.environ.get("ALPHAVANTAGE_API_KEY", "") or os.environ.get("ALPHA_VANTAGE_API_KEY", "") or API_KEY).strip() if not API_KEY: print(f"{YELLOW}⚠ ALPHAVANTAGE_API_KEY is not set (env or .env) — Alpha Vantage " f"requests will fail until it is configured.{RESET}") try: from config import ( OPENAI_API_KEY, GEMINI_API_KEY, ANTHROPIC_API_KEY, KIMI_API_KEY, ) except ImportError: OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "") GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "") KIMI_API_KEY = os.environ.get("KIMI_API_KEY", "") # Per-call LLM network timeout. SDK defaults are ~600s, so a single unreachable provider hangs # a debate round for 10 minutes before giving up; 120s fails fast on network trouble while # leaving ample headroom for a legitimately slow multi-round response. _LLM_TIMEOUT = 120.0 _LLM_SYSTEM_PROMPT = """You are an expert quantitative equity analyst. You are given the outputs of multiple quantitative models that predict a stock's 1-month return. Your job is to synthesize these signals, identify conflicts, assess which sources are most credible for this specific stock, and produce a FINAL predicted price and a BUY/HOLD/SELL recommendation. Key principles: - ML models extrapolate from historical patterns and can be wildly wrong for stocks in unusual regimes (parabolic runs, near-bankruptcy, post-M&A) - Technical signals are most reliable for mean-reversion in normal regimes, less so for momentum crashes - Fundamental signals are most reliable for mature companies, less so for growth/speculative names - Insider selling is one of the strongest bearish signals, especially when unanimous (0 buys) - When analyst targets diverge strongly from ML predictions, investigate why — analysts have forward info the model doesn't - Be skeptical of extreme predictions (>10% monthly return). Most stocks don't move that much. - Your confidence should be LOW when sources strongly disagree Recommendation guidelines: - STRONG BUY: High-conviction upside >5% in 1 month with strong fundamental + technical alignment - BUY: Expected upside >2% with reasonable confidence - HOLD: Expected return -2% to +2%, or high uncertainty - SELL: Expected downside >2% with reasonable confidence - STRONG SELL: High-conviction downside >5% with multiple bearish signals confirming You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _LLM_USER_TEMPLATE = """Analyze {ticker} ({months}-month forecast) and synthesize these quantitative signals: CURRENT: ${current_price:.2f} {signals_text} Respond with ONLY this JSON (no markdown): {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "<2-3 sentences explaining your synthesis and which signals you weighted most>", "risk_flags": [""], "override_ml": }}""" _BEAR_SYSTEM_PROMPT = """You are a senior risk manager at a top-tier hedge fund. Your mandate is capital preservation. Your role: make the STRONGEST POSSIBLE BEAR CASE for why this stock will underperform or decline. - Stress-test every bullish assumption. Assume quant models are overfit to recent momentum. - Focus on: downside risks, valuation traps, insider selling (what was traded, never why), technical exhaustion, macro headwinds. - NEVER recommend STRONG BUY. If risk/reward is genuinely exceptional, the best you can say is HOLD. - Be specific — every claim must cite an actual number from the data provided. - Look beyond the financials: if a SMART-MONEY FLOW block (13F super-investors / insider Form-4) or RECENT NEWS HEADLINES block is present, you MUST address it — what positions and trades do the filings show (they record holdings, never intent or motive), are the "catalysts" hype? If it cuts against your thesis, concede it explicitly. - Build a COMPREHENSIVE thesis, not just a numbers read. Explicitly weigh: (a) business & growth outlook — demand trend, TAM, product cycle, guidance credibility; (b) competitive landscape & moat — share loss, pricing power erosion, disruptors, substitution; (c) business/operational risks — customer/geographic concentration, regulation, litigation, execution, balance-sheet fragility; (d) management & capital allocation — track record, dilution, buybacks at high prices, incentive alignment. Ground each in the earnings-call excerpt, news, and fundamentals trends provided plus well-established facts about the company — never fabricate specifics you can't support. - A missed risk is a career-ending failure. A missed opportunity cost is noise. Recommendation guidelines (from your perspective): - STRONG SELL: high-conviction downside >5%, multiple confirming signals - SELL: expected downside >2% with reasonable confidence - HOLD: risk/reward is balanced but tilted bearish; not worth adding You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _BEAR_USER_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month {signals_text} As the RISK MANAGER, make the strongest possible bear case. In "key_points", include ONLY the topics that are material (drop any you have nothing specific for); each point states its conclusion first, then the number that proves it (1-2 sentences). Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "", "key_points": [ {{"topic": "Business outlook", "point": ""}}, {{"topic": "Competitive landscape", "point": ""}}, {{"topic": "Business risks", "point": ""}}, {{"topic": "Management", "point": ""}}, {{"topic": "Valuation", "point": ""}}, {{"topic": "Smart money & news", "point": "<13F/insider positioning; real catalyst vs hype>"}} ], "risk_flags": [""], "override_ml": }}""" _BULL_SYSTEM_PROMPT = """You are a growth equity analyst at a top-tier asset manager. Your mandate is alpha generation. Your role: make the STRONGEST POSSIBLE BULL CASE for why this stock will outperform. - Challenge every bearish assumption. Assume the market is underpricing the growth trajectory. - Focus on: momentum catalysts, earnings power, changes in reported institutional positions, sector tailwinds, quality metrics. - NEVER recommend STRONG SELL. If the setup is genuinely terrible, the worst you say is HOLD. - Be specific — every claim must cite an actual number from the data provided. - Look beyond the financials: if a SMART-MONEY FLOW block (13F super-investors / insider Form-4) or RECENT NEWS HEADLINES block is present, you MUST address it — name the tracked investors whose reported positions rose (positions, never intent), cite real catalysts. If it cuts against your thesis, concede it explicitly. - Build a COMPREHENSIVE thesis, not just a numbers read. Explicitly weigh: (a) business & growth outlook — demand drivers, TAM expansion, product/roadmap momentum, operating leverage; (b) competitive landscape & moat — durable share gains, pricing power, network effects, switching costs; (c) key business risks and why they are manageable or already priced in; (d) management & capital allocation — execution track record, founder/insider alignment, smart reinvestment or buybacks. Ground each in the earnings-call excerpt, news, and fundamentals trends provided plus well-established facts about the company — never fabricate specifics you can't support. - A missed 10% move is a failure. A 2% drawdown is noise. Recommendation guidelines (from your perspective): - STRONG BUY: high-conviction upside >5% in 1 month with strong alignment - BUY: expected upside >2% with reasonable confidence - HOLD: opportunity exists but timing or entry is not yet ideal You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _BULL_USER_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month {signals_text} As the GROWTH ANALYST, make the strongest possible bull case. In "key_points", include ONLY the topics that are material (drop any you have nothing specific for); each point states its conclusion first, then the number that proves it (1-2 sentences). Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "", "key_points": [ {{"topic": "Business outlook", "point": ""}}, {{"topic": "Competitive landscape", "point": ""}}, {{"topic": "Business risks", "point": ""}}, {{"topic": "Management", "point": ""}}, {{"topic": "Valuation", "point": ""}}, {{"topic": "Smart money & news", "point": "<13F/insider accumulation; real catalysts>"}} ], "risk_flags": [""], "override_ml": }}""" _PM_BLIND_SYSTEM_PROMPT = """You are a portfolio manager responsible for a long-only equity book. Make an initial independent assessment of this stock based purely on the quantitative data provided. Be decisive — HOLD should reflect genuine balance, not uncertainty avoidance. Every claim must rest on a specific figure from the data, written as its [[TOKEN]], never typed. You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _PM_BLIND_USER_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month {signals_text} As the PORTFOLIO MANAGER, make your initial independent assessment before hearing from your analysts. Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "<2-3 sentences: your synthesis and key conviction drivers>", "risk_flags": [""], "override_ml": }}""" _PM_SYNTHESIS_SYSTEM_PROMPT = """You are a portfolio manager. You have heard from your risk manager (bear case) and growth analyst (bull case). Your task: assign probability weights to each scenario — NOT pick a side outright. Rules: - p_bear + p_bull + p_neutral = 1.0 (p_neutral = 1 - p_bear - p_bull) - p_bear must be between 0.10 and 0.65 (you cannot dismiss the bear entirely or let it dominate) - p_bull must be between 0.10 and 0.65 (same constraint) - The final expected return will be computed MECHANICALLY from your probabilities — do NOT engineer a target number - Your reasoning is the decision thesis: state what you concluded and the evidence that decided it — not a recap of each analyst Do NOT simply average. Make a genuine judgment about which case is better supported by the data. You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _PM_SYNTHESIS_USER_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month YOUR INITIAL VIEW (made before hearing analysts): Return: {pm_r1_return:+.2f}% | Recommendation: {pm_r1_rec} | Direction: {pm_r1_direction} Reasoning: {pm_r1_reasoning} RISK MANAGER (Bear) says: Return: {bear_return:+.2f}% → ${bear_price:.2f} | Recommendation: {bear_rec} | Confidence: {bear_conf:.0f}% Thesis: {bear_reasoning} Key Risks: {bear_risks} GROWTH ANALYST (Bull) says: Return: {bull_return:+.2f}% → ${bull_price:.2f} | Recommendation: {bull_rec} | Confidence: {bull_conf:.0f}% Thesis: {bull_reasoning} Key Catalysts: {bull_catalysts} Assign probability weights to each scenario. Constraints: p_bear ∈ [0.10, 0.65], p_bull ∈ [0.10, 0.65], p_neutral = 1 - p_bear - p_bull ≥ 0.05 Respond with ONLY this JSON: {{ "p_bear": , "p_bull": , "p_neutral": , "conviction": , "reasoning": "<2-3 sentences: which specific arguments moved your probabilities and why>", "risk_flags": [""], "override_ml": }}""" _NEUTRAL_SYSTEM_PROMPT = """You are an independent equity research analyst. Your mandate is objectivity — no bull or bear bias. Synthesize the quantitative signals and report what the data actually shows. Give equal weight to bullish and bearish evidence. Do not advocate. Do not anchor to any prior view. Be specific — every claim must cite an actual number from the data. Do not stop at valuation and financials: if SMART-MONEY FLOW (13F super-investor / insider Form-4 positioning) or RECENT NEWS HEADLINES are present in the data, weigh them alongside the fundamentals and say what they show. Cover the whole business objectively: (a) business & growth outlook, (b) competitive landscape & moat, (c) key business/operational risks, (d) management & capital allocation — grounding each in the earnings-call excerpt, news, and fundamentals trends plus well-established facts about the company, without fabricating specifics you can't support. Recommendation guidelines (apply them neutrally): - STRONG BUY: overwhelming multi-signal bullish confluence, low downside risk - BUY: majority of signals bullish with manageable risk - HOLD: mixed signals, high uncertainty, or balanced risk/reward - SELL: majority of signals bearish with confirmed downside risk - STRONG SELL: overwhelming multi-signal bearish confluence You MUST respond with valid JSON only. No markdown, no explanation outside JSON.""" _NEUTRAL_USER_TEMPLATE = """Analyze {ticker} ({months}-month forecast) and provide an objective, unbiased assessment of these quantitative signals: CURRENT: ${current_price:.2f} {signals_text} As the NEUTRAL ANALYST, report what the data shows without advocacy. In "key_points", include ONLY the topics that are material (drop any you have nothing specific for); each point is 1-2 sentences citing actual numbers. Keep "reasoning" to a single-sentence net call. Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "", "key_points": [ {{"topic": "Business outlook", "point": ""}}, {{"topic": "Competitive landscape", "point": ""}}, {{"topic": "Business risks", "point": ""}}, {{"topic": "Management", "point": ""}}, {{"topic": "Valuation", "point": ""}}, {{"topic": "Smart money & news", "point": ""}} ], "risk_flags": [""], "override_ml": }}""" _BEAR_REBUTTAL_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month NEUTRAL ANALYST VIEW (objective baseline): Return: {neutral_return:+.2f}% | Recommendation: {neutral_rec} | Confidence: {neutral_conf:.0f}% Reasoning: {neutral_reasoning} YOUR BEAR POSITION (Round 2): Return: {my_return:+.2f}% → ${my_price:.2f} | Recommendation: {my_rec} Thesis: {my_reasoning} GROWTH ANALYST (Bull) counters with: Return: {bull_return:+.2f}% → ${bull_price:.2f} | Recommendation: {bull_rec} | Confidence: {bull_conf:.0f}% Thesis: {bull_reasoning} Key Catalysts: {bull_catalysts} Rebut ONLY the specific points where you disagree with the bull — do NOT restate your whole case. List the 2-3 sharpest contested points. Each key_point: "topic" = the bull's claim you're challenging (quote it briefly), "point" = your one-sentence counter with a hard number. Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "", "key_points": [ {{"topic": "", "point": ""}}, {{"topic": "", "point": ""}} ], "risk_flags": [""], "override_ml": , "changed_view": }}""" _BULL_REBUTTAL_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month NEUTRAL ANALYST VIEW (objective baseline): Return: {neutral_return:+.2f}% | Recommendation: {neutral_rec} | Confidence: {neutral_conf:.0f}% Reasoning: {neutral_reasoning} YOUR BULL POSITION (Round 2): Return: {my_return:+.2f}% → ${my_price:.2f} | Recommendation: {my_rec} Thesis: {my_reasoning} RISK MANAGER (Bear) counters with: Return: {bear_return:+.2f}% → ${bear_price:.2f} | Recommendation: {bear_rec} | Confidence: {bear_conf:.0f}% Thesis: {bear_reasoning} Key Risks: {bear_risks} Rebut ONLY the specific points where you disagree with the bear — do NOT restate your whole case. List the 2-3 sharpest contested points. Each key_point: "topic" = the bear's claim you're challenging (quote it briefly), "point" = your one-sentence counter with a hard number. Respond with ONLY this JSON: {{ "expected_return_pct": , "predicted_price": , "recommendation": "", "confidence": , "direction": "", "reasoning": "", "key_points": [ {{"topic": "", "point": ""}}, {{"topic": "", "point": ""}} ], "risk_flags": [""], "override_ml": , "changed_view": }}""" _PM_SYNTHESIS_USER_TEMPLATE = """STOCK: {ticker} | CURRENT: ${current_price:.2f} | HORIZON: {months}-month NEUTRAL ANALYST VIEW (objective baseline — no advocacy): Return: {neutral_return:+.2f}% | Recommendation: {neutral_rec} | Confidence: {neutral_conf:.0f}% Reasoning: {neutral_reasoning} RISK MANAGER (Bear) final position after debate: Return: {bear_return:+.2f}% → ${bear_price:.2f} | Recommendation: {bear_rec} | Confidence: {bear_conf:.0f}% Thesis: {bear_reasoning} Key Risks: {bear_risks} GROWTH ANALYST (Bull) final position after debate: Return: {bull_return:+.2f}% → ${bull_price:.2f} | Recommendation: {bull_rec} | Confidence: {bull_conf:.0f}% Thesis: {bull_reasoning} Key Catalysts: {bull_catalysts} Assign probability weights to each scenario. The neutral view is your anchor — use it to judge which advocate is better supported by the data. Constraints: p_bear ∈ [0.10, 0.65], p_bull ∈ [0.10, 0.65], p_neutral = 1 - p_bear - p_bull ≥ 0.05 Respond with ONLY this JSON: {{ "p_bear": , "p_bull": , "p_neutral": , "conviction": , "reasoning": "<2-3 sentences: which specific arguments moved your probabilities, referencing the neutral baseline>", "risk_flags": [""], "override_ml": }}""" def _dedup_flags(flags): seen, out = set(), [] for f in flags: k = f.lower().strip() if k not in seen: seen.add(k) out.append(f) return out def _repair_truncated_json(text): """Best-effort close of a truncated JSON object (the one containing p_bear). Tracks the open-bracket STACK so closers are appended in correct nesting order. Returns a dict or None.""" start = text.find('"p_bear"') if start == -1: return None open_at = text.rfind("{", 0, start) if open_at == -1: return None frag = text[open_at:] # candidate cut points: the full fragment, then progressively earlier safe points cuts = [len(frag)] for tok in ('"}', '},', '"]', '",', '}', '"'): i = frag.rfind(tok) if i > 0: cuts.append(i + len(tok)) for c in cuts: body = frag[:c].rstrip().rstrip(",") stack = []; in_str = False; esc = False for ch in body: if in_str: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': in_str = False continue if ch == '"': in_str = True elif ch in "{[": stack.append(ch) elif ch in "}]": if stack: stack.pop() if in_str: body += '"' # a dangling key ("topic": ) or trailing colon can't be closed — drop it body = re.sub(r',?\s*"[^"]*"\s*:\s*$', "", body) body = body.rstrip().rstrip(",") while stack: body += "}" if stack.pop() == "{" else "]" try: d = json.loads(body) if isinstance(d, dict) and "p_bear" in d: return d except Exception: continue return None def _parse_pm_synthesis(raw, tokens_in, tokens_out): """Parse PM synthesis JSON — different schema from standard analyst responses.""" cleaned = raw if "```" in cleaned: cleaned = re.sub(r"```(?:json)?\s*", "", cleaned) cleaned = cleaned.strip() try: data = json.loads(cleaned) except json.JSONDecodeError: data = None # Tolerate a preamble/trailing prose around the object, and nested braces # (the flat [^{}] regex fails when reasoning contains braces): scan for the # first balanced {...} that mentions p_bear. start = cleaned.find('"p_bear"') if start != -1: open_at = cleaned.rfind("{", 0, start) if open_at != -1: depth = 0 for i in range(open_at, len(cleaned)): if cleaned[i] == "{": depth += 1 elif cleaned[i] == "}": depth -= 1 if depth == 0: try: data = json.loads(cleaned[open_at:i + 1]) except Exception: data = None break if data is None: # TRUNCATION REPAIR: a long synthesis can hit max_tokens mid-JSON (seen on PLTR: # the raw head held valid p_bear/p_bull/p_neutral but the tail was cut). Throwing # the whole thing away collapses the capstone to an equal-weight blend, so try to # close the object: drop the trailing partial token, then append the closers the # nesting still needs. If the scalar fields survive, the synthesis is usable even # when the last key_point is lost. data = _repair_truncated_json(cleaned) if data is None: return None return { "p_bear": float(np.clip(_num(data.get("p_bear")) if _num(data.get("p_bear")) is not None else 0.33, 0.10, 0.65)), "p_bull": float(np.clip(_num(data.get("p_bull")) if _num(data.get("p_bull")) is not None else 0.33, 0.10, 0.65)), "p_neutral": float(np.clip(_num(data.get("p_neutral")) if _num(data.get("p_neutral")) is not None else 0.34, 0.05, 0.80)), "conviction": float(np.clip(_num(data.get("conviction")) if _num(data.get("conviction")) is not None else 0.5, 0.1, 1.0)), "reasoning": str(data.get("reasoning", "") or (data.get("thesis") if isinstance(data.get("thesis"), str) else "")), "key_points": _coerce_key_points(data.get("key_points")), "structured": _structured_fields(data), "tipping_point": (str(data.get("tipping_point"))[:300] if data.get("tipping_point") else None), # structured flip conditions travel with the synthesis: a fixed key list here was silently # dropping them, which is what forced the renderer back onto free-prose tipping points "triggers": (data.get("triggers") if isinstance(data.get("triggers"), list) else []), # evidence lineage and any quantitative tilt: dropped here before, which left the evidence # ledger and the scenario-probability decomposition with no input in production "evidence_used": [str(x) for x in (data.get("evidence_used") or []) if isinstance(x, (str, int))][:12], "quant_tilt": ({k: float(np.clip(float(v), -0.2, 0.2)) for k, v in data["quant_tilt"].items() if k in ("bear", "neutral", "bull") and isinstance(v, (int, float))} if isinstance(data.get("quant_tilt"), dict) else {}), "risk_flags": list(data.get("risk_flags", [])), "override_ml": bool(data.get("override_ml", False)), "tokens_in": tokens_in, "tokens_out": tokens_out, } def _call_claude_synthesis(user_msg, verbose=False): """Call Claude for PM synthesis — uses a higher token budget.""" if not ANTHROPIC_API_KEY: return None import anthropic client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY, timeout=_LLM_TIMEOUT) response = claude_create(client, model="claude-sonnet-5", max_tokens=4500, # long reasoning truncated the JSON (PLTR); parser now also repairs truncation system=_PM_SYNTHESIS_SYSTEM_PROMPT, messages=[{"role": "user", "content": user_msg}], temperature=1.0, ) raw = "".join(b.text for b in response.content if getattr(b, "type", "") == "text").strip() if verbose: print(f" {DIM}Claude synthesis tokens: {response.usage.input_tokens} in / {response.usage.output_tokens} out{RESET}") parsed = _parse_pm_synthesis(raw, response.usage.input_tokens, response.usage.output_tokens) if parsed is None: # Surface WHY the capstone failed instead of silently collapsing to equal-weight. print(f" {DIM}PM parse failed; raw head: {raw[:160].replace(chr(10), ' ')}{RESET}", flush=True) return parsed # ─── 3-Panel Signal Judge (scan Stage 4 — PASS/FLAG/REJECT; matches weekly pattern) ─ _JUDGE_SYSTEM_ROUND1_MONTHLY = """You are an expert equity risk analyst evaluating a QUANTITATIVE 1-month trade signal for a human operator. Your job is NOT to predict returns. You assess whether the quantitative signal is trustworthy and assign a CONFIDENCE ADJUSTMENT — a percentage by which you would modify the quant model's pick confidence. Key calibration for 1-MONTH horizon: - Earnings within 10 days is a RISK FACTOR that widens uncertainty. It is NOT an automatic rejection. Many stocks drift predictably in the 10-20 days before earnings (pre-earnings drift). Only reject if earnings combine with multiple OTHER severe risks. - RSI 65-75 is normal momentum, not overbought. RSI 75-85 is elevated. RSI > 85 is extreme. - A stock selected by a momentum scanner WILL have positive momentum. Don't penalize the trait that selected it — instead check whether that momentum is sustainable or exhausted. - Positive directional accuracy (>55%) and signal stability (>30%) from walk-forward are STRONG positive signals. Weight these heavily. - For 1-month horizon, fundamental quality matters more than at 1-week. Check if high expected returns are supported by fundamentals. - Macro regime (rising rates, widening credit spreads) can impair momentum strategies — flag if detected. Your confidence adjustment should be: +5 to +15: Signal looks clean, risks are minimal → boosts pick confidence -5 to -15: Minor risks present but signal is tradeable → slight reduction -15 to -30: Significant risks, trade with reduced size → moderate reduction -30 to -50: Severe risks, likely better to wait → strong reduction Below -50: Critical risks, do not trade → near-rejection REJECT is reserved for: data quality issues, circuit breaker / halted stock, known fraud / SEC investigation, or signal that is clearly broken (e.g., dir_acc 20% on high confidence). Respond with valid JSON only. No markdown.""" _JUDGE_USER_ROUND1_MONTHLY = """CANDIDATE PROFILE (1-month horizon): {ticker} @ ${current_price:.2f} Blended expected return: {blend_ret_pct:+.2f}% Walk-forward prediction (OOS mean): {wf_pred_pct:+.2f}% Direction: {direction} | Confidence: {confidence:.0f}% Risk score (0-1, higher=worse): {risk_score:.2f} Pick confidence (walk-forward): {pick_confidence:.1%} Dir accuracy (historical windows): {dir_accuracy:.1%} Signal stability: {signal_stability:.1%} Walk-forward Rank IC: {rank_ic:.3f} OOS samples: {oos_samples} P(return>5% in 21d): {p5:.1%} | P(return>0%): {ppos:.1%} Earnings: {earn_str} History: {history_days} days of price data Key technicals: RSI {rsi:.0f} | BB%B {bb:.2f} | Mom1m {mom1m:+.2f}% {extra_context} Respond with ONLY this JSON: {{ "confidence_adjustment_pct": , "verdict": "", "severity": , "risk_flags": ["", "..."], "reasoning": "<1-2 sentences>", "confidence": }}""" _JUDGE_USER_DEBATE_MONTHLY = """STOCK: {ticker} YOUR ROUND-1 POSITION: Verdict: {my_verdict} | Adjustment: {my_adjustment:+d}% | Severity: {my_severity}/5 Reasoning: {my_reasoning} OTHER ANALYSTS: {other_positions} Respond with ONLY this JSON: {{ "confidence_adjustment_pct": , "verdict": "", "severity": , "risk_flags": ["..."], "reasoning": "<1-2 sentences>", "confidence": }}""" # ─── Monthly LLM Selector (single-call, all candidates) ────────────────────── _SELECTOR_SYSTEM_MONTHLY = """You are an expert portfolio selector evaluating quantitative 1-month trade candidates. Your role is to assess SIGNAL RELIABILITY — not to re-predict price direction (the ML ensemble already does that). VERDICT CRITERIA: PASS — signal is reliable; no disqualifying risk flags FLAG — tradeable but with elevated risk (e.g. earnings proximity, overbought RSI, low OOS samples); reduce size REJECT — signal is unreliable or risk is unacceptable ADJUSTMENT: integer −50 to +20 scaling the pick_confidence. Negative = penalty (FLAG/REJECT). Positive = modest boost (rare; only high-conviction PASS). SEVERITY: 1 (minor concern) to 5 (disqualifying) for the worst risk flag. 1-MONTH horizon calibration: - RSI 65-75 is normal momentum; RSI >85 is extreme. Don't penalise the trait that selected it. - Earnings within 10 days is a risk factor — NOT auto-rejection. Only reject if combined with multiple other severe risks. - Positive dir_acc >55% and stability >30% from walk-forward are STRONG positives. Weight heavily. - High prob_gt_5pct (>15%) combined with strong dir_acc is a quality signal. - rank_ic > 0.20 indicates good cross-sectional predictability. - For 1-month horizon, fundamental quality matters more than at 1-week. When calibration data is provided, use it to anchor severity. Respond ONLY with valid JSON. No markdown fences, no explanation outside the JSON.""" _SELECTOR_USER_TEMPLATE_MONTHLY = """SCAN DATE: {scan_date} {calibration_context} MACRO CONTEXT (use to adjust sector/leverage risk — do not over-weight): {macro_context} MONTHLY CANDIDATES ({n} total, ordered by pick_confidence descending): {candidates_text} Return JSON exactly: {{ "assessments": [ {{ "ticker": "", "verdict": "PASS|FLAG|REJECT", "adjustment": <-50 to +20>, "severity": <1-5>, "risk_flags": ["concise flag", ...], "reasoning": "" }} ], "ranked": ["ticker1", "ticker2", ...], "best_pick": "", "best_pick_reasoning": "" }}""" # ─── Monthly scan outcome tracker ───────────────────────────────────────────── # Logs every Stage-3-validated pick → resolves actual 21d return after 35 days # → feeds per-confidence-bucket calibration stats into the Stage 4 selector. _SCAN_PICK_LOG_MONTHLY = CACHE_DIR / "scan_pick_log_monthly.jsonl" # ─── LLM Call Functions ────────────────────────────────────────────────────── def _call_openai(user_msg, system=None, verbose=False): """Call OpenAI GPT-5.5.""" if not OPENAI_API_KEY: print(f" {YELLOW}OPENAI_API_KEY not set{RESET}") return None from openai import OpenAI client = OpenAI(api_key=OPENAI_API_KEY, timeout=_LLM_TIMEOUT) response = client.chat.completions.create( model="gpt-5.1", messages=[ {"role": "system", "content": system or _LLM_SYSTEM_PROMPT}, {"role": "user", "content": user_msg}, ], temperature=1.0, max_completion_tokens=4000, ) raw = response.choices[0].message.content.strip() tokens_in = response.usage.prompt_tokens tokens_out = response.usage.completion_tokens if verbose: print(f" {DIM}GPT-5.5 tokens: {tokens_in} in / {tokens_out} out{RESET}") return _parse_llm_response(raw, tokens_in, tokens_out, "gpt-5.1") def _call_kimi(user_msg, system=None, verbose=False): """Call Moonshot AI Kimi K2.6 with thinking-mode fallback (matches weekly / signal judge).""" if not KIMI_API_KEY: print(f" {YELLOW}KIMI_API_KEY not set{RESET}") return None from openai import OpenAI import time as _time # Dedicated shorter timeout: Kimi/Moonshot thinking-mode can stall, and this caller # retries internally — a tighter per-request bound keeps the debate from freezing. client = OpenAI(api_key=KIMI_API_KEY, base_url="https://api.moonshot.ai/v1", timeout=75.0) def _try_once(extra_body=None, temperature=1.0, max_completion_tokens=4096): kwargs = dict( model="kimi-k2.6", messages=[ {"role": "system", "content": system or _LLM_SYSTEM_PROMPT}, {"role": "user", "content": user_msg}, ], temperature=temperature, max_completion_tokens=max_completion_tokens, ) if extra_body is not None: try: return client.chat.completions.create(**kwargs, extra_body=extra_body) except TypeError: pass except Exception as e: err = str(e) if "thinking" in err.lower() or "400" in err: pass else: raise return client.chat.completions.create(**kwargs) max_retries = 3 _deadline = _time.monotonic() + 200.0 # hard wall-clock cap — never stall the debate for attempt in range(max_retries): if _time.monotonic() > _deadline: print(f" {DIM}Kimi (Bear) time budget exhausted — skipping{RESET}", flush=True) return None if attempt > 0: print(f" {DIM}Kimi (Bear) retry {attempt+1}/{max_retries}...{RESET}", flush=True) try: response = None last_err = None for cfg in ( # kimi-k2.6 enforces mode-specific temperature: thinking-DISABLED requires # temp 0.6, thinking-ENABLED requires temp 1.0. cfg1 previously used 0.3 → # rejected with HTTP 400 on every call, forcing the slow thinking path (cfg2) # and stalling / dropping the Bear. cfg1 at 0.6 makes the fast path work (~10s). {"extra_body": {"thinking": {"type": "disabled"}}, "temperature": 0.6, "max_completion_tokens": 4096}, {"extra_body": None, "temperature": 1.0, "max_completion_tokens": 6144}, ): try: response = _try_once( extra_body=cfg["extra_body"], temperature=cfg["temperature"], max_completion_tokens=cfg["max_completion_tokens"], ) break except Exception as e: last_err = e err = str(e) if "429" in err: raise continue if response is None: raise last_err if last_err else RuntimeError("Kimi: no response") msg = response.choices[0].message raw = _moonshot_assistant_text(msg) tokens_in = response.usage.prompt_tokens if response.usage else 0 tokens_out = response.usage.completion_tokens if response.usage else 0 finish = getattr(response.choices[0], "finish_reason", None) if verbose: print( f" {DIM}Kimi K2.6 tokens: {tokens_in} in / {tokens_out} out (finish: {finish}){RESET}" ) if not raw or not raw.strip(): if attempt < max_retries - 1: _time.sleep(min(3 * (attempt + 1), 8)) continue return None raw = raw.strip() result = _parse_llm_response(raw, tokens_in, tokens_out, "kimi-k2.6") if result is None and attempt < max_retries - 1: _time.sleep(min(3 * (attempt + 1), 8)) continue return result except Exception as e: err_str = str(e) if "401" in err_str: raise Exception("Kimi K2.6: invalid/expired API key (401)") if "429" in err_str: wait = min(10 * (2 ** attempt), 30) if attempt >= 2: print(f" {DIM}Kimi rate limited — skipping after {attempt+1} attempts{RESET}") return None print(f" {DIM}Kimi rate limited, retrying in {wait}s...{RESET}") _time.sleep(wait) continue if attempt < max_retries - 1: _time.sleep(min(4 * (attempt + 1), 10)) continue return None return None def _call_gemini(user_msg, system=None, verbose=False): """Call Google Gemini Flash 2.5 Lite (fast/cheap for scanning).""" if not GEMINI_API_KEY: print(f" {YELLOW}GEMINI_API_KEY not set{RESET}") return None import google.generativeai as genai genai.configure(api_key=GEMINI_API_KEY) safety_off = [ {"category": c, "threshold": "BLOCK_NONE"} for c in ["HARM_CATEGORY_HARASSMENT", "HARM_CATEGORY_HATE_SPEECH", "HARM_CATEGORY_SEXUALLY_EXPLICIT", "HARM_CATEGORY_DANGEROUS_CONTENT"] ] model = genai.GenerativeModel( "gemini-3-flash-preview-lite", system_instruction=system or _LLM_SYSTEM_PROMPT, ) response = model.generate_content( user_msg, generation_config=genai.GenerationConfig(temperature=0.3, max_output_tokens=2200), safety_settings=safety_off, request_options={"timeout": _LLM_TIMEOUT}, ) raw = response.text.strip() tokens_in = response.usage_metadata.prompt_token_count if hasattr(response, 'usage_metadata') else 0 tokens_out = response.usage_metadata.candidates_token_count if hasattr(response, 'usage_metadata') else 0 if verbose: print(f" {DIM}Gemini Flash tokens: {tokens_in} in / {tokens_out} out{RESET}") return _parse_llm_response(raw, tokens_in, tokens_out, "gemini-3-flash-preview-lite") def _call_gemini_3_pro(user_msg, system=None, verbose=False): """Call Google Gemini 3 Flash — used as Neutral Analyst in Round 1.""" if not GEMINI_API_KEY: print(f" {YELLOW}GEMINI_API_KEY not set{RESET}") return None import google.generativeai as genai genai.configure(api_key=GEMINI_API_KEY) safety_off = [ {"category": c, "threshold": "BLOCK_NONE"} for c in ["HARM_CATEGORY_HARASSMENT", "HARM_CATEGORY_HATE_SPEECH", "HARM_CATEGORY_SEXUALLY_EXPLICIT", "HARM_CATEGORY_DANGEROUS_CONTENT"] ] # gemini-3-flash-preview: production-stable; uses internal thinking so needs high max_output_tokens model = genai.GenerativeModel( "gemini-3-flash-preview", system_instruction=system or _LLM_SYSTEM_PROMPT, ) response = model.generate_content( user_msg, generation_config=genai.GenerationConfig(temperature=0.3, max_output_tokens=8192), safety_settings=safety_off, request_options={"timeout": _LLM_TIMEOUT}, ) raw = response.text.strip() tokens_in = response.usage_metadata.prompt_token_count if hasattr(response, "usage_metadata") else 0 tokens_out = response.usage_metadata.candidates_token_count if hasattr(response, "usage_metadata") else 0 if verbose: print(f" {DIM}Gemini 3 Flash tokens: {tokens_in} in / {tokens_out} out{RESET}") print(f" {DIM}Gemini raw: {raw[:200]}{RESET}") parsed = _parse_llm_response(raw, tokens_in, tokens_out, "gemini-3-flash-preview") if parsed is None and not verbose: import sys print(f" {YELLOW}Gemini parse failed. raw[:400]={raw[:400]!r}{RESET}", file=sys.stderr) return parsed def _call_claude(user_msg, system=None, verbose=False): """Call Anthropic Claude Opus 4.7.""" if not ANTHROPIC_API_KEY: print(f" {YELLOW}ANTHROPIC_API_KEY not set{RESET}") return None import anthropic client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY, timeout=_LLM_TIMEOUT) response = claude_create(client, model="claude-sonnet-5", max_tokens=2000, system=system or _LLM_SYSTEM_PROMPT, messages=[{"role": "user", "content": user_msg}], temperature=1.0, ) raw = response.content[0].text.strip() tokens_in = response.usage.input_tokens tokens_out = response.usage.output_tokens if verbose: print(f" {DIM}Claude Opus tokens: {tokens_in} in / {tokens_out} out{RESET}") return _parse_llm_response(raw, tokens_in, tokens_out, "claude-sonnet-5") def _coerce_key_points(kp): """Normalize an analyst's key_points into a clean [{topic, point}] list. Tolerates dicts, bare strings, or 'Topic: text' strings; returns [] if unusable so the UI falls back to the prose reasoning (no regression).""" out = [] if isinstance(kp, list): for it in kp[:8]: if isinstance(it, dict): topic = str(it.get("topic", "") or "").strip() point = str(it.get("point", it.get("text", "")) or "").strip() elif isinstance(it, str) and it.strip(): s = it.strip() topic, sep, rest = s.partition(":") topic, point = (topic.strip(), rest.strip()) if (sep and len(topic) < 40) else ("", s) else: continue if point: def _trunc(t, n): # keep whole sentences only: break at the last sentence end inside the limit. Authored prose is # never cut mid-sentence (a cut sentence is a different, unaudited claim); a single over-long # first sentence is kept whole and left to prose ingestion if len(t) <= n: return t ends = [m.end() for m in re.finditer(r"[.!?](?=\s)", t[:n])] if ends: return t[:ends[-1]].strip() first = re.search(r"[.!?](?=\s|$)", t) return t[:first.end()].strip() if first else t out.append({"topic": _trunc(topic, 72), "point": _trunc(point, 600)}) return out def _parse_llm_response(raw, tokens_in, tokens_out, model_name): """Parse LLM JSON response into prediction dict.""" # Strip markdown fences if present cleaned = raw if "```" in cleaned: cleaned = re.sub(r"```(?:json)?\s*", "", cleaned) cleaned = cleaned.strip() try: data = json.loads(cleaned) except json.JSONDecodeError as _jde: _parse_err = str(_jde) # Find the outermost { ... } block that contains expected_return_pct. # This handles models that prepend plain text before the JSON. data = None pos = 0 while pos < len(cleaned): start = cleaned.find('{', pos) if start == -1: break depth, i = 0, start while i < len(cleaned): if cleaned[i] == '{': depth += 1 elif cleaned[i] == '}': depth -= 1 if depth == 0: candidate = cleaned[start:i + 1] if '"expected_return_pct"' in candidate: try: data = json.loads(candidate) except json.JSONDecodeError: pass break i += 1 if isinstance(data, dict): break pos = start + 1 if not isinstance(data, dict): # Last resort: a truncated-but-well-formed prefix (max_tokens cut) can usually be # closed by the bracket-stack repairer used for the PM synthesis. try: data = _repair_truncated_json(cleaned) except Exception: data = None if not isinstance(data, dict): import sys print(f" {YELLOW} json_err={getattr(locals(), '_parse_err', '?')!r} cleaned[:200]={cleaned[:200]!r}{RESET}", file=sys.stderr) return None _er_raw = data.get("expected_return_pct") has_return = _num(_er_raw) is not None expected_return = (_num(_er_raw) or 0.0) / 100.0 _cf = _num(data.get("confidence")) _cf = 0.5 if _cf is None else (_cf / 100.0 if _cf > 1.0 else _cf) confidence = min(max(_cf, 0.1), 1.0) direction = data.get("direction", "NEUTRAL") reasoning = data.get("reasoning", "") or (data.get("thesis") if isinstance(data.get("thesis"), str) else "") risk_flags = data.get("risk_flags", []) override_ml = data.get("override_ml", False) changed_view = data.get("changed_view", False) predicted_price = data.get("predicted_price") recommendation = data.get("recommendation", "HOLD") key_points = _coerce_key_points(data.get("key_points")) # Normalize recommendation if recommendation: recommendation = recommendation.upper().strip() valid_recs = {"STRONG BUY", "BUY", "HOLD", "SELL", "STRONG SELL"} if recommendation not in valid_recs: recommendation = "HOLD" return { "expected_return": expected_return, "confidence": confidence, "direction": direction, "reasoning": reasoning, "key_points": key_points, "risk_flags": risk_flags, "override_ml": override_ml, "changed_view": changed_view, "predicted_price": predicted_price, "recommendation": recommendation, "model": model_name, "tokens_in": tokens_in, "tokens_out": tokens_out, # structured-argument fields (crucible.args/2) pass through untouched for engine/arguments.py; a reply # without a numeric expected_return_pct is marked so a rebuttal keeps its opening numbers "structured": _structured_fields(data), "has_return": has_return, } def _num(x): """A number from a JSON scalar ("-12.5", "−12.5%", 12) or None.""" if isinstance(x, bool): return None if isinstance(x, (int, float)): return float(x) if isinstance(x, str): t = x.strip().replace("\u2212", "-").replace("%", "").replace(",", "") try: return float(t) except ValueError: return None return None def _structured_fields(data): """The structured-argument keys of a persona reply, copied as JSON (validated downstream, never here).""" import copy as _copy out = {k: _copy.deepcopy(data[k]) for k in ("thesis", "thesis_rests_on", "points", "stance", "answers", "factors", "summary") if k in data} kp = data.get("key_points") if "points" not in out and isinstance(kp, list) and any(isinstance(x, dict) and ("view" in x or "kind" in x) for x in kp): out["points"] = _copy.deepcopy(kp) return out # ─── Final Blend & Display ─────────────────────────────────────────────────── # ─── Main ───────────────────────────────────────────────────────────────────── if __name__ == "__main__": main()