import json import os import re from dotenv import load_dotenv # Load environment variables from .venv file load_dotenv(".venv") # ── Gemini via REST (no SDK) ─────────────────────────────── try: import requests as _requests GEMINI_AVAILABLE = True except ImportError: _requests = None GEMINI_AVAILABLE = False GEMINI_MODEL = "gemini-2.5-flash" GEMINI_URL = ( "https://generativelanguage.googleapis.com/v1beta/models/" "{model}:generateContent" ) # Injected into the prompt so the model can give specific, grounded explanations rather than generic RL platitudes. TICKER_CONTEXT = { "NVDA": ( "NVDA experienced an extreme bull run in 2023–2024 driven by " "AI infrastructure demand. Its price appreciation exceeded 200% " "during this period — a regime completely unlike the 2015–2022 " "training window where NVDA was a volatile but range-bound " "semiconductor stock." ), "GOOGL": ( "GOOGL underwent a post-pandemic valuation reset followed by a " "strong recovery in 2023–2024. The test period coincided with " "renewed AI optimism around Google DeepMind and Gemini, which " "created momentum patterns different from the training window." ), "AAPL": ( "AAPL exhibited a selective, patient strategy with very few trades " "and long average hold durations (~145 days). The test period " "included Apple's Vision Pro launch and continued services growth, " "creating a steady uptrend that rewarded holding." ), "MSFT": ( "MSFT was the strongest performer. The test period coincided with " "Azure AI growth and the OpenAI partnership driving consistent " "institutional buying. The agent learned a moderate-frequency " "trading strategy that captured most of this uptrend." ), "AMZN": ( "AMZN recovered strongly in 2023–2024 after a severe 2022 drawdown. " "The agent learned an aggressive strategy with high trade frequency " "and 60% position sizing, effectively catching the recovery wave." ), } # ── Training period context ──────────────────────────────────── TRAINING_CONTEXT = ( "The PPO agents were trained on data from 2015 to mid 2026 using " "a custom Gymnasium environment with a 30-day lookback window, " "MultiDiscrete action space (direction × size), and a Differential " "Sharpe Ratio reward function with idle and invalid action penalties. " "Hyperparameters were tuned via Optuna TPE with 30 trials per ticker, " "optimising validation Sharpe Ratio." ) def build_analysis_prompt( ticker: str, kpis_rl: dict, kpis_bnh: dict, kpis_sma: dict, actions: list, trades: int, period: str, ) -> str: # Action breakdown percentages total = len(actions) if actions else 1 hold_pct = actions.count(0) / total * 100 if actions else 0 buy_pct = actions.count(1) / total * 100 if actions else 0 sell_pct = actions.count(2) / total * 100 if actions else 0 ticker_ctx = TICKER_CONTEXT.get(ticker, "No specific context available.") prompt = f""" You are a quantitative finance analyst reviewing the performance of a Reinforcement Learning trading agent (PPO algorithm) on {ticker} stock. === TRAINING CONTEXT === {TRAINING_CONTEXT} === TICKER-SPECIFIC CONTEXT === {ticker_ctx} === BACKTEST RESULTS ({period} live data) === RL Agent (PPO): Final Value : {kpis_rl.get("Final Value", "N/A")} Total Return : {kpis_rl.get("Total Return", "N/A")} Ann. Return : {kpis_rl.get("Ann. Return", "N/A")} Ann. Volatility : {kpis_rl.get("Ann. Volatility", "N/A")} Sharpe Ratio : {kpis_rl.get("Sharpe Ratio", "N/A")} Sortino Ratio : {kpis_rl.get("Sortino Ratio", "N/A")} Max Drawdown : {kpis_rl.get("Max Drawdown", "N/A")} Calmar Ratio : {kpis_rl.get("Calmar Ratio", "N/A")} Total Trades : {trades} Action Breakdown : Hold {hold_pct:.1f}% | Buy {buy_pct:.1f}% | Sell {sell_pct:.1f}% Buy & Hold: Total Return : {kpis_bnh.get("Total Return", "N/A")} Sharpe Ratio : {kpis_bnh.get("Sharpe Ratio", "N/A")} Max Drawdown : {kpis_bnh.get("Max Drawdown", "N/A")} SMA Crossover: Total Return : {kpis_sma.get("Total Return", "N/A")} Sharpe Ratio : {kpis_sma.get("Sharpe Ratio", "N/A")} Max Drawdown : {kpis_sma.get("Max Drawdown", "N/A")} === YOUR TASK === Provide a structured analysis in EXACTLY this JSON format. Return ONLY the JSON object, no markdown, no preamble: {{ "summary": "2-3 sentence executive summary of overall performance", "strengths": [ "Specific strength 1 with numbers from the results", "Specific strength 2", "Specific strength 3" ], "weaknesses": [ "Specific weakness 1 — diagnose the root cause", "Specific weakness 2", "Specific weakness 3" ], "failure_diagnosis": "1-2 paragraphs specifically explaining WHY the agent performed the way it did on {ticker}. Reference the training period, regime change, reward function, action space, or hyperparameters as appropriate. Be precise and technical.", "improvements": [ "Concrete improvement 1 with implementation detail", "Concrete improvement 2", "Concrete improvement 3", "Concrete improvement 4" ], "verdict": "Final verdict in this exact form: start with a letter grade (A/B/C/D/F) based on actual performance, then state the alpha vs Buy & Hold and the Sharpe with real numbers, then a clear deployment call. Grade A = clearly beat B&H with Sharpe >= 1; B = small positive edge; C = roughly matched B&H; D = underperformed; F = never traded. Be specific and non-cliché — no 'shows promise but needs tuning'." }} """ return prompt.strip() def get_ai_analysis( ticker: str, kpis_rl: dict, kpis_bnh: dict, kpis_sma: dict, actions: list, trades: int, period: str, ) -> dict: if _requests is None: return _fallback_analysis(ticker, kpis_rl, kpis_bnh, trades) try: api_key = os.getenv("GEMINI_API_KEY") if not api_key: return _fallback_analysis( ticker, kpis_rl, kpis_bnh, trades, note="GEMINI_API_KEY not found in environment.", ) prompt = build_analysis_prompt( ticker, kpis_rl, kpis_bnh, kpis_sma, actions, trades, period ) # ── Gemini REST call (no SDK, no websockets dependency) ─────── url = GEMINI_URL.format(model=GEMINI_MODEL) resp = _requests.post( url, headers={"Content-Type": "application/json"}, params={"key": api_key}, json={"contents": [{"parts": [{"text": prompt}]}]}, timeout=30, ) # Handle HTTP errors WITHOUT raise_for_status: its message embeds the # full request URL, which includes ?key=... and would leak the API key. if resp.status_code != 200: msg = { 429: "rate limit reached — try again shortly", 503: "service temporarily unavailable — try again shortly", 500: "service error — try again shortly", 403: "request rejected (check API key configuration)", 400: "bad request", }.get(resp.status_code, f"HTTP {resp.status_code}") return _fallback_analysis( ticker, kpis_rl, kpis_bnh, trades, note=f"AI analysis unavailable: {msg}.", ) data = resp.json() # Extract text from the first candidate's first part. raw_text = "" try: raw_text = data["candidates"][0]["content"]["parts"][0]["text"] except (KeyError, IndexError, TypeError): raw_text = "" if not raw_text: raise ValueError( "Gemini returned an empty response. This usually happens due " "to safety filters or an invalid API key." ) raw_text = raw_text.strip() # Strip markdown code fences if present raw_text = re.sub(r"^```(?:json)?", "", raw_text).strip() raw_text = re.sub(r"```$", "", raw_text).strip() analysis = json.loads(raw_text) # Ensure a grade exists for the medallion, derived from real KPIs, # so the AI path renders the same performance-graded verdict UI. if "grade" not in analysis: analysis["grade"] = _compute_grade(kpis_rl, kpis_bnh, trades) analysis["score"] = _compute_score(kpis_rl, kpis_bnh, trades) return analysis except json.JSONDecodeError: # Response was not valid JSON — extract what we can return _fallback_analysis( ticker, kpis_rl, kpis_bnh, trades, note="AI response could not be parsed." ) except Exception as e: # Defensive: scrub anything that looks like an API key or the request # URL from the exception text before showing it, so a key can never # leak into the UI even from an unexpected error path. safe = re.sub(r"key=[\w\-]+", "key=***", str(e)) safe = re.sub(r"https?://\S+", "[endpoint]", safe) # Keep it short and generic safe = safe[:120] return _fallback_analysis( ticker, kpis_rl, kpis_bnh, trades, note=f"AI analysis unavailable: {safe}", ) def _compute_score(kpis_rl: dict, kpis_bnh: dict, trades: int) -> int: def parse_pct(v): try: return float(str(v).replace("%", "").replace("+", "").strip()) except Exception: return 0.0 rl_ret = parse_pct(kpis_rl.get("Total Return", "0%")) bnh_ret = parse_pct(kpis_bnh.get("Total Return", "0%")) rl_sharpe = parse_pct(kpis_rl.get("Sharpe Ratio", "0")) alpha = rl_ret - bnh_ret if trades == 0: return 0 # Alpha component: 0 at -20%, 50 at +20% (capped) alpha_score = max(0.0, min(50.0, (alpha + 20.0) / 40.0 * 50.0)) # Sharpe component: 0 at <=0, 40 at >=2.0 (capped) sharpe_score = max(0.0, min(40.0, rl_sharpe / 2.0 * 40.0)) # Participation: up to 10 for a sane (non-degenerate) trade count part_score = 10.0 if 2 <= trades <= 80 else (5.0 if trades > 0 else 0.0) return int(round(alpha_score + sharpe_score + part_score)) def _compute_grade(kpis_rl: dict, kpis_bnh: dict, trades: int) -> str: """Grade the agent A–F from real KPIs. Single source of truth for both the AI and fallback paths.""" def parse_pct(v): try: return float(str(v).replace("%", "").replace("+", "").strip()) except Exception: return 0.0 rl_ret = parse_pct(kpis_rl.get("Total Return", "0%")) bnh_ret = parse_pct(kpis_bnh.get("Total Return", "0%")) rl_sharpe = parse_pct(kpis_rl.get("Sharpe Ratio", "0")) alpha = rl_ret - bnh_ret if trades == 0: return "F" if alpha >= 5.0 and rl_sharpe >= 1.0: return "A" if alpha >= 0.0 and rl_sharpe >= 0.5: return "B" if alpha >= -5.0: return "C" return "D" def _fallback_analysis( ticker: str, kpis_rl: dict, kpis_bnh: dict, trades: int, note: str = "", ) -> dict: def parse_pct(v): try: return float(str(v).replace("%", "").replace("+", "").strip()) except Exception: return 0.0 rl_ret = parse_pct(kpis_rl.get("Total Return", "0%")) bnh_ret = parse_pct(kpis_bnh.get("Total Return", "0%")) rl_sharpe = parse_pct(kpis_rl.get("Sharpe Ratio", "0")) rl_dd = parse_pct(kpis_rl.get("Max Drawdown", "0%")) outperform = rl_ret > bnh_ret ticker_ctx = TICKER_CONTEXT.get(ticker, "") summary = ( f"The PPO agent on {ticker} achieved a total return of " f"{kpis_rl.get('Total Return', 'N/A')} vs Buy & Hold's " f"{kpis_bnh.get('Total Return', 'N/A')} over the evaluation period. " f"{'The agent outperformed the passive benchmark.' if outperform else 'The agent underperformed the passive benchmark.'}" ) strengths = [] # A non-participating agent (zero trades) has no genuine strengths — its # flat 0% drawdown and untouched capital are artifacts of inaction, not # skill. Only credit strengths when the agent actually traded. if trades > 0: if trades > 20: strengths.append( f"Active participation: {trades} trades show the agent engaged with the market rather than ignoring it." ) if rl_sharpe > 0.5: strengths.append( f"Positive Sharpe Ratio of {kpis_rl.get('Sharpe Ratio')} indicates the agent generated risk-adjusted returns above the risk-free rate." ) if outperform: strengths.append( f"Beat Buy & Hold by {rl_ret - bnh_ret:.1f}% — the policy added value over passively holding the asset." ) if rl_dd > -25: strengths.append( f"Controlled drawdown of {kpis_rl.get('Max Drawdown')} suggests the reward function's drawdown penalty was effective." ) if not strengths: if trades == 0: strengths.append( "None — the agent never entered the market, so there is no behaviour to credit. The flat capital and zero drawdown reflect inaction, not risk management." ) else: strengths.append( "The agent traded but produced no clear edge; no individual metric stands out as a genuine strength." ) weaknesses = [] if rl_ret < bnh_ret: weaknesses.append( f"Underperformed Buy & Hold by {bnh_ret - rl_ret:.1f}% — the added complexity did not translate to returns." ) if rl_sharpe < 0: weaknesses.append( "Negative Sharpe Ratio means the strategy's volatility was not compensated by returns." ) if trades == 0: weaknesses.append( "Zero executed trades indicates a degenerate policy — the agent learned to always attempt invalid actions." ) weaknesses.append( "Limited out-of-sample generalisation due to the 2015–2022 training window not capturing recent market regimes." ) failure_diagnosis = ( f"The agent's performance on {ticker} reflects a distribution shift between " f"the training period (2015–2022) and the test period. {ticker_ctx} " f"The PPO agent optimises for a Differential Sharpe Ratio reward which " f"penalises volatility — in a strongly trending market this causes the agent " f"to exit positions too early, missing sustained directional moves that a " f"simple Buy & Hold captures fully." ) improvements = [ "Retrain with data through 2026 to include recent market regimes in the training distribution.", "Increase the entropy coefficient (ent_coef) to 0.05+ to prevent premature policy collapse.", "Add a regime detection module (HMM or volatility clustering) to switch strategies dynamically.", "Implement online learning or periodic retraining to adapt to distribution shift in production.", ] # ── Performance-graded verdict (driven by actual numbers) ───────── # Grade on three axes: alpha vs Buy & Hold, risk-adjusted return (Sharpe), # and whether the agent actually traded coherently. No clichés — the # verdict states the grade, the numbers behind it, and the deployment call. alpha = rl_ret - bnh_ret if trades == 0: grade, headline, deploy = ( "F", f"The agent never traded on {ticker} — it sat in cash for the entire window and returned {rl_ret:+.1f}% against Buy & Hold's {bnh_ret:+.1f}%.", "Not deployable. A non-participating policy has nothing to deploy.", ) elif alpha >= 5.0 and rl_sharpe >= 1.0: grade, headline, deploy = ( "A", f"The agent beat Buy & Hold by {alpha:+.1f}% on {ticker} ({rl_ret:+.1f}% vs {bnh_ret:+.1f}%) at a Sharpe of {rl_sharpe:.2f}, across {trades} trades.", "Promising enough to paper-trade forward on unseen data before any live capital.", ) elif alpha >= 0.0 and rl_sharpe >= 0.5: grade, headline, deploy = ( "B", f"The agent edged Buy & Hold by {alpha:+.1f}% on {ticker} ({rl_ret:+.1f}% vs {bnh_ret:+.1f}%) at a Sharpe of {rl_sharpe:.2f}, but the margin is thin.", "Not ready for live capital — the edge is too small to survive real costs and slippage. Validate across more regimes first.", ) elif alpha >= -5.0: grade, headline, deploy = ( "C", f"The agent tracked Buy & Hold within {abs(alpha):.1f}% on {ticker} ({rl_ret:+.1f}% vs {bnh_ret:+.1f}%) — it is effectively replicating the benchmark, not beating it.", "Not worth deploying over simply holding the asset, which is cheaper and simpler.", ) else: grade, headline, deploy = ( "D", f"The agent underperformed Buy & Hold by {abs(alpha):.1f}% on {ticker} ({rl_ret:+.1f}% vs {bnh_ret:+.1f}%) at a Sharpe of {rl_sharpe:.2f}.", "Do not deploy. The added complexity actively cost return versus just holding.", ) verdict = f"{headline} {deploy}" result = { "summary": summary, "strengths": strengths, "weaknesses": weaknesses, "failure_diagnosis": failure_diagnosis, "improvements": improvements, "verdict": verdict, "grade": grade, "score": _compute_score(kpis_rl, kpis_bnh, trades), } if note: result["note"] = note return result def format_analysis_as_html(analysis: dict, ticker: str, theme: dict) -> str: # ── Claude identity palette (warm, editorial) ───────────────────── ink = "#1F1E1C" # near-black warm ink muted = "#6E6B65" # secondary text clay = "#C96442" # the one bold accent — Claude's signature rust ivory = "#FAF9F5" # page surface cream = "#F0EEE6" # card surface line = "#E3DFD3" # hairline warm border on_clay = "#FFFFFF" grn = "#3D8C5F" # desaturated to sit in the warm palette amb = "#C7901F" red = "#BC4633" # Map legacy names used below primary, secondary, tertiary = ink, muted, clay neutral, surface, on_primary = ivory, cream, on_clay SERIF = "'IBM Plex Sans',sans-serif" SANS = "'IBM Plex Sans',sans-serif" MONO = "'IBM Plex Mono',monospace" # Performance score 0–100 → accent color + label score = int(analysis.get("score", 0)) if score >= 75: grade_color, grade_label = grn, "Strong" elif score >= 55: grade_color, grade_label = grn, "Marginal edge" elif score >= 40: grade_color, grade_label = amb, "Matches benchmark" elif score > 0: grade_color, grade_label = red, "Underperforms" else: grade_color, grade_label = red, "No participation" def md(t: str) -> str: if not isinstance(t, str): return t t = re.sub(r"\*\*(.*?)\*\*", r"\1", t) t = re.sub(r"\*(.*?)\*", r"\1", t) t = re.sub( r"`(.*?)`", f"\\1", t, ) return t def icon(name: str) -> str: return { "cpu": "◆", "calendar": "", "message-square": "", "trending-up": "↑", "alert-triangle": "△", "bug": "◈", "flowchart": "", "target": "", "chevron-right": "→", "check": "✓", "x": "✕", }.get(name, "") def pattern(kind: str, color: str) -> str: """Light geometric SVG pattern, absolutely positioned, very low opacity. Sits behind card content as a subtle texture.""" pid = f"p{kind}{abs(hash(kind+color))%9999}" defs = { "grid": f'', "dots": f'', "diag": f'', "tri": f'', }.get(kind, "") return ( f'' ) def eyebrow(text: str, color: str) -> str: return ( f"

{text}

" ) def card( title: str, content: str, icon_name: str, color: str, subtitle: str = "" ) -> str: return f"""
{pattern("grid", color)}
{eyebrow(title, color)} {content} {"

' + subtitle + "

" if subtitle else ""}
""" def get_icon_name(color: str) -> str: if color == grn: return "check" elif color == amb: return "alert-triangle" else: return "x" def bullet_list(items: list, color: str) -> str: icon_char = icon(get_icon_name(color)) lis = "".join( f"
  • " f'{icon_char}' f"{md(item)}
  • " for item in items ) return f'' def flow_section(title: str, items: list) -> str: nodes = "" for i, item in enumerate(items): nodes += f"""
    {i + 1}

    {md(item)}

    """ return f"""
    {pattern("diag", clay)}
    {eyebrow("How to improve it", clay)} {nodes}
    """ note_html = "" if "note" in analysis: note_html = f"

    {analysis['note']}

    " strengths_html = bullet_list(analysis.get("strengths", []), grn) weaknesses_html = bullet_list(analysis.get("weaknesses", []), amb) improvements_flow = flow_section("How to improve it", analysis.get("improvements", [])) grade_medallion = f"""
    {score} / 100
    """ html = f"""

    Agent performance review

    {ticker} · PPO trading agent

    Reviewed by Claude
    {pattern("dots", grade_color)}
    {grade_medallion}

    Performance score · {grade_label}

    {md(analysis.get("verdict", ""))}

    {pattern("grid", clay)}
    {eyebrow("What happened", clay)}

    {md(analysis.get("summary", ""))}

    {pattern("dots", grn)}
    {eyebrow("What worked", grn)} {strengths_html}
    {pattern("tri", amb)}
    {eyebrow("What held it back", amb)} {weaknesses_html}
    {pattern("diag", red)}
    {eyebrow("Why it behaved this way", red)}

    {md(analysis.get("failure_diagnosis", ""))}

    {improvements_flow} {note_html}

    This is research analysis, not financial advice. Validate on unseen data before risking capital.

    """ return html