#!/usr/bin/env python3 """ Shadow LLM Evaluator Runs daily at 8am ET before market open. Compares shadow vs real bot performance, generates report. Pushes report to GitHub shadow_backup/ for Jake + Claude to review. """ import json, os, glob, requests, base64, urllib.request from datetime import datetime BASE = "/opt/services/shadow_llm" SHADOW = "/opt/services/bots/shadow_bot/Stock_Bot" REAL_BOT = "/opt/services/bots/trading_bot/Stock_Bot" OLLAMA = "http://localhost:11434/api/chat" MODEL = "deepseek-r1:14b" PAT = open("/opt/services/bots/trading_bot/.env").read().split("GITHUB_PAT=")[1].split()[0] if os.path.exists("/opt/services/bots/trading_bot/.env") else "" REPO = "Jake-Culberson/Claud-Code" def load_metrics(bot_dir): """Load performance metrics from a bot directory.""" metrics = {"status": "no_data", "source": bot_dir} try: state = json.load(open(f"{bot_dir}/logs/agent_state.json")) agents = [(k,v) for k,v in state.items() if not k.startswith('__')] total_pnl = sum(v.get('total_pnl',0) for _,v in agents) total_trades = sum(v.get('trades',0) for _,v in agents) total_wins = sum(v.get('wins',0) for _,v in agents) win_rate = total_wins / max(total_trades,1) * 100 metrics = { "total_pnl": round(total_pnl, 2), "total_trades": total_trades, "win_rate_pct": round(win_rate, 1), "n_agents": len(agents), "avg_pnl_per_trade": round(total_pnl / max(total_trades,1), 4), "top_3": sorted( [(k, round(v.get('total_pnl',0),2)) for k,v in agents], key=lambda x: x[1], reverse=True )[:3], } except FileNotFoundError: metrics = {"status": "fresh_start"} except Exception as e: metrics = {"status": f"error: {e}"} return metrics def load_proposals(): """Load pending and applied proposals.""" proposals = {"pending": [], "applied": [], "rejected": []} for path in sorted(glob.glob(f"{BASE}/proposals/proposal_*.json")): try: p = json.load(open(path)) status = p.get("status","unknown") item = { "title": p.get("title","?"), "change": p.get("change",{}), "confidence": p.get("confidence","?"), "path": path, } if status == "pending_review": proposals["pending"].append(item) elif status == "applied": proposals["applied"].append(item) elif "rejected" in status: proposals["rejected"].append(item) except: pass return proposals def generate_comparison_report(): """Generate daily comparison report.""" shadow = load_metrics(SHADOW) real = load_metrics(REAL_BOT) proposals = load_proposals() ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Ask LLM to analyze the comparison prompt = f""" Analyze this comparison between the shadow (test) bot and real bot: REAL BOT: {json.dumps(real, indent=2)} SHADOW BOT: {json.dumps(shadow, indent=2)} PENDING PROPOSALS (not yet applied): {len(proposals['pending'])} proposals APPLIED PROPOSALS: {json.dumps(proposals['applied'], indent=2)} Write a brief, honest comparison report in plain English. Include: 1. Which bot is performing better and by how much 2. Whether applied proposals are helping 3. Which pending proposals look most promising 4. Any concerns or anomalies in the data 5. Recommendation for Jake on what to do next Be direct. If both bots have no trades yet (fresh start), say so clearly. Do not hallucinate metrics. Only report what the data shows. Keep it under 300 words. """ try: r = requests.post(OLLAMA, json={ "model": MODEL, "messages": [{"role": "user", "content": prompt}], "stream": False, "options": {"temperature": 0.1} }, timeout=300) analysis = r.json().get("message",{}).get("content","Analysis unavailable") if "" in analysis: analysis = analysis.split("")[-1].strip() except Exception as e: analysis = f"AI analysis unavailable: {e}" report = { "generated_at": ts, "real_bot": real, "shadow_bot": shadow, "proposals": proposals, "ai_analysis": analysis, } # Save locally report_path = f"{BASE}/reports/report_{datetime.now().strftime('%Y%m%d')}.json" json.dump(report, open(report_path,'w'), indent=2) # Also write human-readable version readable = f"""# Shadow vs Real Bot — Daily Comparison Generated: {ts} ## Performance | Metric | Real Bot | Shadow Bot | |---|---|---| | Net P&L | ${real.get('total_pnl', 'N/A'):+} | ${shadow.get('total_pnl', 'N/A'):+} | | Total Trades | {real.get('total_trades', 'N/A')} | {shadow.get('total_trades', 'N/A')} | | Win Rate | {real.get('win_rate_pct', 'N/A')}% | {shadow.get('win_rate_pct', 'N/A')}% | ## AI Analysis {analysis} ## Pending Proposals ({len(proposals['pending'])}) """ for p in proposals['pending']: readable += f"- **{p['title']}** (confidence: {p['confidence']})\n" readable += f" Change: {p['change'].get('parameter')} → {p['change'].get('new_value')}\n" readable += f" File: {p['path']}\n\n" readable_path = f"{BASE}/reports/report_{datetime.now().strftime('%Y%m%d')}.md" open(readable_path,'w').write(readable) # Push to GitHub if PAT: push_to_github(readable_path, readable, "Stock_Bot/shadow_backup/daily_report.md") push_to_github(report_path, json.dumps(report, indent=2), "Stock_Bot/shadow_backup/daily_report.json") print(readable) return report def push_to_github(local_path, content, remote_path): """Push a file to GitHub.""" API = f"https://api.github.com/repos/{REPO}/contents" headers = {"Authorization": f"Bearer {PAT}", "Content-Type": "application/json"} try: req = urllib.request.Request(f"{API}/{remote_path}", headers={"Authorization": f"Bearer {PAT}"}) sha = "" try: with urllib.request.urlopen(req) as resp: sha = json.loads(resp.read()).get("sha","") except: pass pl = { "message": f"shadow: daily report {datetime.now().strftime('%Y-%m-%d')}", "content": base64.b64encode(content.encode()).decode() } if sha: pl["sha"] = sha req2 = urllib.request.Request(f"{API}/{remote_path}", data=json.dumps(pl).encode(), method="PUT", headers=headers) with urllib.request.urlopen(req2) as resp: print(f"Pushed: {remote_path}") except Exception as e: print(f"GitHub push error: {e}") if __name__ == "__main__": generate_comparison_report()