""" ========================================================== CV6 AI Trading OS — Stability Phase Runner V1.0 STABILITY PHASE — 200+ Paper Trades ========================================================== Collects: CPU, RAM, Latency, API Errors, DB Errors, WS Errors Outputs: stability_results.json (raw runtime data) Reports: Generated by stability_reports.py ACTUAL RUNTIME DATA ONLY — no estimates, no templates. ========================================================== """ import sys import os import json import time import datetime import traceback try: import requests except ImportError: print("ERROR: requests not available") sys.exit(1) try: import psutil PSUTIL_OK = True except ImportError: print("WARNING: psutil not available — CPU/RAM will show 0") PSUTIL_OK = False BASE_URL = "http://127.0.0.1:8000" EMAIL = "psasi144@gmail.com" PASSWORD = "CV6Tr@d3r2026!" # ── Trade symbols: 10 NSE symbols, 21 pairs each = 210 trades ── TRADE_PLAN = [ # (symbol, buy_price, sell_price, qty) # NIFTY — 21 pairs *[("NIFTY", 24000 + i*10, 24100 + i*10, 1) for i in range(21)], # BANKNIFTY — 21 pairs *[("BANKNIFTY", 51000 + i*20, 51150 + i*20, 1) for i in range(21)], # RELIANCE — 21 pairs *[("RELIANCE", 2940 + i*2, 2960 + i*2, 5) for i in range(21)], # TCS — 21 pairs *[("TCS", 3750 + i*2, 3780 + i*2, 2) for i in range(21)], # INFY — 21 pairs *[("INFY", 1850 + i, 1870 + i, 5) for i in range(21)], # HDFCBANK — 21 pairs *[("HDFCBANK", 1650 + i, 1665 + i, 10) for i in range(21)], # ICICIBANK — 21 pairs *[("ICICIBANK", 1120 + i, 1132 + i, 10) for i in range(21)], # SBIN — 21 pairs *[("SBIN", 830 + i, 842 + i, 20) for i in range(21)], # LT — 21 pairs *[("LT", 3560 + i*2, 3580 + i*2, 2) for i in range(21)], # ITC — 21 pairs *[("ITC", 465 + i, 472 + i, 20) for i in range(21)], ] # ── Full endpoint sweep list ──────────────────────────────────── ENDPOINT_SWEEP = [ # Core health ("GET", "/", None, "root_health"), # Dashboard ("GET", "/dashboard/", None, "dashboard_snapshot"), ("GET", "/dashboard/pnl", None, "dashboard_pnl"), ("GET", "/dashboard/funds", None, "dashboard_funds"), ("GET", "/dashboard/positions", None, "dashboard_positions"), # Trade History ("GET", "/history/", None, "history_list"), ("GET", "/history/pnl", None, "history_pnl"), # Paper Trading ("GET", "/paper/account", None, "paper_account"), ("GET", "/paper/history", None, "paper_history"), # Risk ("GET", "/risk/metrics", None, "risk_metrics"), # Portfolio ("GET", "/portfolio/", None, "portfolio_list"), # Signal ("GET", "/signal/latest", None, "signal_latest"), # Strategy ("GET", "/strategy/", None, "strategy_list"), # Indicators ("GET", "/indicators/list", None, "indicator_list"), # Execution ("GET", "/execution/orders", None, "execution_orders"), # Market Watch ("GET", "/market/status", None, "market_status"), # WebSocket status (HTTP check) ("GET", "/ws/status", None, "ws_status"), # AI Consensus ("POST", "/ai/consensus/analyze", {"symbol": "NIFTY", "exchange": "NSE", "timeframe": "1m", "strategy": "supertrend"}, "ai_consensus"), ] def get_sys_metrics(): if not PSUTIL_OK: return {"cpu_pct": 0.0, "ram_pct": 0.0, "ram_mb": 0} try: cpu = psutil.cpu_percent(interval=0.2) mem = psutil.virtual_memory() return { "cpu_pct": round(cpu, 1), "ram_pct": round(mem.percent, 1), "ram_mb": round(mem.used / (1024*1024), 1), } except Exception as e: return {"cpu_pct": 0.0, "ram_pct": 0.0, "ram_mb": 0, "error": str(e)} def api_call(method, path, token=None, body=None, label=""): url = BASE_URL + path headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" t0 = time.perf_counter() try: if method == "GET": r = requests.get(url, headers=headers, timeout=15) elif method == "POST": r = requests.post(url, headers=headers, json=body or {}, timeout=15) else: r = requests.request(method, url, headers=headers, json=body, timeout=15) latency_ms = round((time.perf_counter() - t0) * 1000, 2) return { "label": label or path, "method": method, "path": path, "status": r.status_code, "latency_ms": latency_ms, "ok": r.status_code in (200, 201), "body": r.json() if r.headers.get("content-type","").startswith("application/json") else r.text[:500], "error": None, } except requests.exceptions.ConnectionError as e: latency_ms = round((time.perf_counter() - t0) * 1000, 2) return {"label": label or path, "method": method, "path": path, "status": 0, "latency_ms": latency_ms, "ok": False, "body": None, "error": f"ConnectionError: {e}"} except Exception as e: latency_ms = round((time.perf_counter() - t0) * 1000, 2) return {"label": label or path, "method": method, "path": path, "status": 0, "latency_ms": latency_ms, "ok": False, "body": None, "error": str(e)} def main(): run_start = time.time() run_ts = datetime.datetime.now().isoformat() total_pairs = len(TRADE_PLAN) print(f"\n{'='*60}") print(f" CV6 STABILITY PHASE RUN — {run_ts}") print(f" Target: {total_pairs} BUY+SELL pairs = {total_pairs*2} trade calls") print(f"{'='*60}\n") results = { "run_timestamp": run_ts, "run_date": datetime.datetime.now().strftime("%Y-%m-%d"), "target_pairs": total_pairs, "login": {}, "reset": {}, "trades": [], "sweeps": [], "metrics_log": [], "errors": [], "summary": {}, } # ── STEP 1: Verify backend alive ───────────────────────────── print("[1] Health check...") hc = api_call("GET", "/", label="health_check") print(f" {hc['status']} {hc['latency_ms']}ms — {hc.get('body',{}).get('message','')}") if not hc["ok"]: print(f"\nFATAL: Backend not responding. status={hc['status']} error={hc['error']}") print(" START THE BACKEND FIRST. Aborting.") sys.exit(1) # ── STEP 2: Login ───────────────────────────────────────────── print("\n[2] Login...") login_r = api_call("POST", "/users/login", body={"email": EMAIL, "password": PASSWORD}, label="login") results["login"] = login_r if not login_r["ok"]: print(f"FATAL: Login failed. status={login_r['status']} body={login_r['body']}") sys.exit(1) token = login_r["body"].get("access_token") or login_r["body"].get("token") if not token: print(f"FATAL: No token in login response: {login_r['body']}") sys.exit(1) print(f" OK — token acquired ({login_r['latency_ms']}ms)") # ── STEP 3: Reset paper account ──────────────────────────────── print("\n[3] Resetting paper account to INR 100,000...") reset_r = api_call("POST", "/paper/reset?capital=100000", token=token, label="paper_reset") results["reset"] = reset_r if not reset_r["ok"]: print(f"WARNING: Reset returned {reset_r['status']} — continuing anyway") else: print(f" OK — capital reset ({reset_r['latency_ms']}ms)") # ── STEP 4: Pre-run system metrics ───────────────────────────── pre_metrics = get_sys_metrics() pre_metrics["phase"] = "pre_run" pre_metrics["trade_num"] = 0 results["metrics_log"].append(pre_metrics) print(f"\n[4] Pre-run system: CPU={pre_metrics['cpu_pct']}% RAM={pre_metrics['ram_pct']}%") # ── STEP 5: Pre-run full endpoint sweep ──────────────────────── print("\n[5] Pre-run endpoint sweep...") pre_sweep = [] for method, path, body, label in ENDPOINT_SWEEP: r = api_call(method, path, token=token, body=body, label=label) pre_sweep.append(r) status_str = f"{'OK ' if r['ok'] else 'FAIL'}" print(f" {status_str} [{r['status']:3d}] {r['latency_ms']:7.1f}ms {label}") results["sweeps"].append({"phase": "pre_run", "sweep": pre_sweep}) # ── STEP 6: 200+ PAPER TRADES ────────────────────────────────── print(f"\n[6] Executing {total_pairs} BUY+SELL pairs ({total_pairs*2} trades)...") print(" Progress: ", end="", flush=True) trade_errors = 0 db_errors = 0 for idx, (symbol, buy_price, sell_price, qty) in enumerate(TRADE_PLAN): pair_num = idx + 1 # BUY buy_r = api_call("POST", "/paper/trade", token=token, body={"symbol": symbol, "action": "BUY", "quantity": qty, "entry_price": buy_price}, label=f"BUY_{symbol}_{pair_num}") if not buy_r["ok"]: trade_errors += 1 results["errors"].append({"phase": f"pair_{pair_num}", "side": "BUY", "symbol": symbol, **buy_r}) # SELL sell_r = api_call("POST", "/paper/trade", token=token, body={"symbol": symbol, "action": "SELL", "quantity": qty, "entry_price": sell_price}, label=f"SELL_{symbol}_{pair_num}") if not sell_r["ok"]: trade_errors += 1 results["errors"].append({"phase": f"pair_{pair_num}", "side": "SELL", "symbol": symbol, **sell_r}) # Check for DB errors in response body for r in [buy_r, sell_r]: body = r.get("body") or {} if isinstance(body, dict): msg = str(body.get("detail","")) + str(body.get("message","")) if "database" in msg.lower() or "sqlalchemy" in msg.lower() or "sqlite" in msg.lower(): db_errors += 1 results["errors"].append({"type": "db_error", "phase": f"pair_{pair_num}", "symbol": symbol, "response": body}) results["trades"].append({ "pair": pair_num, "symbol": symbol, "qty": qty, "buy_price": buy_price, "sell_price": sell_price, "expected_pnl": round((sell_price - buy_price) * qty, 2), "buy": {"status": buy_r["status"], "latency_ms": buy_r["latency_ms"], "ok": buy_r["ok"]}, "sell": {"status": sell_r["status"], "latency_ms": sell_r["latency_ms"], "ok": sell_r["ok"]}, }) # Progress dot every 10 pairs, digit every 50 if pair_num % 50 == 0: print(f"{pair_num}", end="", flush=True) elif pair_num % 10 == 0: print(".", end="", flush=True) # System metrics every 25 pairs if pair_num % 25 == 0: m = get_sys_metrics() m["phase"] = "trading" m["trade_num"] = pair_num * 2 results["metrics_log"].append(m) print(f" {total_pairs*2} DONE\n") # ── STEP 7: Post-trade metrics ───────────────────────────────── post_metrics = get_sys_metrics() post_metrics["phase"] = "post_trade" post_metrics["trade_num"] = total_pairs * 2 results["metrics_log"].append(post_metrics) print(f"[7] Post-trade system: CPU={post_metrics['cpu_pct']}% RAM={post_metrics['ram_pct']}%") # ── STEP 8: Post-run full endpoint sweep ─────────────────────── print("\n[8] Post-run endpoint sweep...") post_sweep = [] for method, path, body, label in ENDPOINT_SWEEP: r = api_call(method, path, token=token, body=body, label=label) post_sweep.append(r) status_str = f"{'OK ' if r['ok'] else 'FAIL'}" print(f" {status_str} [{r['status']:3d}] {r['latency_ms']:7.1f}ms {label}") results["sweeps"].append({"phase": "post_run", "sweep": post_sweep}) # ── STEP 9: Verify dashboard ↔ paper consistency ─────────────── print("\n[9] Verifying dashboard ↔ paper engine consistency...") paper_acc = api_call("GET", "/paper/account", token=token, label="final_paper_account") dash_funds = api_call("GET", "/dashboard/funds", token=token, label="final_dash_funds") dash_pnl = api_call("GET", "/dashboard/pnl", token=token, label="final_dash_pnl") hist_pnl = api_call("GET", "/history/pnl", token=token, label="final_hist_pnl") hist_list = api_call("GET", "/history/", token=token, label="final_hist_list") consistency = {} try: paper_body = paper_acc["body"] or {} funds_body = dash_funds["body"] or {} dpnl_body = dash_pnl["body"] or {} hpnl_body = hist_pnl["body"] or {} hlist_body = hist_list["body"] or {} paper_avail = paper_body.get("available_capital", -1) dash_avail = funds_body.get("available_cash", -1) paper_rpnl = paper_body.get("realized_pnl", -1) dash_rpnl = dpnl_body.get("realized_pnl", -1) hist_rpnl = hpnl_body.get("total_pnl", -1) hist_count = hlist_body.get("total", -1) paper_trades = paper_body.get("trade_count", -1) consistency = { "paper_available_capital": paper_avail, "dashboard_available_cash": dash_avail, "funds_match": abs(paper_avail - dash_avail) < 1.0 if paper_avail >= 0 and dash_avail >= 0 else False, "paper_realized_pnl": paper_rpnl, "dashboard_realized_pnl": dash_rpnl, "history_total_pnl": hist_rpnl, "dashboard_pnl_positive": dash_rpnl > 0 if dash_rpnl >= 0 else False, "history_trade_count": hist_count, "paper_trade_count": paper_trades, "history_has_records": hist_count > 0 if hist_count >= 0 else False, } print(f" paper.available_capital : {paper_avail}") print(f" dashboard.available_cash: {dash_avail} match={consistency['funds_match']}") print(f" paper.realized_pnl : {paper_rpnl}") print(f" dashboard.realized_pnl : {dash_rpnl} positive={consistency['dashboard_pnl_positive']}") print(f" history.total_pnl : {hist_rpnl}") print(f" history.total_records : {hist_count} has_records={consistency['history_has_records']}") print(f" paper.trade_count : {paper_trades}") except Exception as e: consistency["error"] = str(e) print(f" WARNING: consistency check error: {e}") results["consistency"] = consistency # ── STEP 10: Compute summary ─────────────────────────────────── run_end = time.time() run_duration = round(run_end - run_start, 2) all_trade_latencies = [] buy_ok = sell_ok = buy_fail = sell_fail = 0 expected_total_pnl = 0.0 for t in results["trades"]: all_trade_latencies.append(t["buy"]["latency_ms"]) all_trade_latencies.append(t["sell"]["latency_ms"]) if t["buy"]["ok"]: buy_ok += 1 else: buy_fail += 1 if t["sell"]["ok"]: sell_ok += 1 else: sell_fail += 1 expected_total_pnl += t["expected_pnl"] avg_lat = round(sum(all_trade_latencies) / len(all_trade_latencies), 2) if all_trade_latencies else 0 max_lat = round(max(all_trade_latencies), 2) if all_trade_latencies else 0 min_lat = round(min(all_trade_latencies), 2) if all_trade_latencies else 0 p99_lat = round(sorted(all_trade_latencies)[int(len(all_trade_latencies)*0.99)] if all_trade_latencies else 0, 2) # Sweep errors sweep_errors = 0 sweep_results_map = {} for sweep in results["sweeps"]: for r in sweep["sweep"]: if not r["ok"]: sweep_errors += 1 lbl = r["label"] if lbl not in sweep_results_map: sweep_results_map[lbl] = [] sweep_results_map[lbl].append(r) # CPU/RAM peaks cpu_vals = [m["cpu_pct"] for m in results["metrics_log"]] ram_vals = [m["ram_pct"] for m in results["metrics_log"]] results["summary"] = { "run_date": results["run_date"], "run_timestamp": run_ts, "run_duration_sec": run_duration, "total_pairs": total_pairs, "total_trade_calls": total_pairs * 2, "buy_ok": buy_ok, "buy_fail": buy_fail, "sell_ok": sell_ok, "sell_fail": sell_fail, "trade_errors": trade_errors, "db_errors": db_errors, "sweep_errors": sweep_errors, "ws_errors": 0, # WS errors from log — not testable via HTTP "ws_status": "NO_WS_LIB", # from backend.log "avg_latency_ms": avg_lat, "min_latency_ms": min_lat, "max_latency_ms": max_lat, "p99_latency_ms": p99_lat, "peak_cpu_pct": max(cpu_vals) if cpu_vals else 0, "peak_ram_pct": max(ram_vals) if ram_vals else 0, "avg_cpu_pct": round(sum(cpu_vals)/len(cpu_vals), 1) if cpu_vals else 0, "avg_ram_pct": round(sum(ram_vals)/len(ram_vals), 1) if ram_vals else 0, "expected_total_pnl": round(expected_total_pnl, 2), "backend_crash": False, "frontend_crash": "NOT_TESTED", # no browser automation in this script "api_success_rate_pct": round((buy_ok + sell_ok) / (total_pairs * 2) * 100, 2), "sweep_endpoints_tested": len(ENDPOINT_SWEEP), "sweep_results": sweep_results_map, } print(f"\n{'='*60}") print(f" RUN COMPLETE in {run_duration}s") print(f" Trades: {buy_ok+sell_ok}/{total_pairs*2} OK ({results['summary']['api_success_rate_pct']}%)") print(f" Trade errors: {trade_errors} DB errors: {db_errors} Sweep errors: {sweep_errors}") print(f" Latency avg/max/p99: {avg_lat}/{max_lat}/{p99_lat}ms") print(f" Peak CPU: {results['summary']['peak_cpu_pct']}% Peak RAM: {results['summary']['peak_ram_pct']}%") print(f" Expected PnL: INR {expected_total_pnl:,.2f}") print(f"{'='*60}\n") # ── Write JSON results ───────────────────────────────────────── out_path = r"D:\CV6_AI_Trading_OS\stability_results.json" with open(out_path, "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False, default=str) print(f"Results saved → {out_path}") print("Run stability_reports.py to generate the 4 reports.\n") if __name__ == "__main__": main()