""" CV6 Live Application QA Check Checks: Backend health, all API endpoints, file integrity, syntax validation Run: python CV6_QA_CHECK.py """ import ast import json import os import sys import time import urllib.request import urllib.error from datetime import datetime BACKEND = "http://127.0.0.1:8000" REPORT = [] PASS = 0 FAIL = 0 WARN = 0 def p(msg): print(msg) REPORT.append(msg) def section(title): p(f"\n{'='*60}") p(f" {title}") p(f"{'='*60}") def check(label, ok, detail="", warn=False): global PASS, FAIL, WARN if ok: status = "✅ PASS" PASS += 1 elif warn: status = "⚠️ WARN" WARN += 1 else: status = "❌ FAIL" FAIL += 1 line = f" {status} {label}" if detail: line += f" → {detail}" p(line) return ok def api_get(path, timeout=5): """GET request to backend. Returns (status_code, body_dict_or_None).""" try: url = BACKEND + path req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=timeout) as r: body = json.loads(r.read().decode()) return r.status, body except urllib.error.HTTPError as e: return e.code, None except Exception as ex: return 0, str(ex) # ─── START ──────────────────────────────────────────────────────────────────── p(f"\n{'#'*60}") p(f" CV6 AI TRADING OS — LIVE QA CHECK") p(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") p(f"{'#'*60}") # ═══════════════════════════════════════════════════════════════ # SECTION 1: BACKEND HEALTH # ═══════════════════════════════════════════════════════════════ section("1. BACKEND HEALTH (localhost:8000)") status, body = api_get("/") backend_up = status == 200 check("Backend running", backend_up, f"HTTP {status}" if status else "NOT RUNNING") if backend_up: check("Root response OK", "success" in str(body) or "CV6" in str(body), str(body)[:60]) # ═══════════════════════════════════════════════════════════════ # SECTION 2: CORE API ENDPOINTS # ═══════════════════════════════════════════════════════════════ section("2. CORE API ENDPOINTS") endpoints = [ ("/docs", "FastAPI Swagger UI"), ("/dashboard/summary", "Dashboard Summary"), ("/autonomous/status", "Autonomous Engine Status"), ("/autonomous/config", "Autonomous Config"), ("/autonomous/scan-results","Smart Scanner Results"), ("/autonomous/capital-pools","Capital Pools"), ("/autonomous/positions", "Open Positions"), ("/autonomous/activity-log","Activity Log"), ("/risk/status", "Risk Guard Status"), ("/risk/config", "Risk Guard Config"), ("/fno/status", "F&O Engine Status"), ("/learning/stats", "Learning Engine Stats"), ("/learning/suggestions", "AI Suggestions"), ("/history/trades", "Trade History"), ("/settings/brokers", "Broker Settings"), ("/settings/ai", "AI Model Settings"), ("/market/watchlist", "Market Watchlist"), ("/alerts/list", "Alerts List"), ] for path, label in endpoints: s, b = api_get(path) ok = s in (200, 201) warn = s in (401, 403, 404) check(label, ok, f"HTTP {s}", warn=warn and not ok) # ═══════════════════════════════════════════════════════════════ # SECTION 3: AUTONOMOUS ENGINE API # ═══════════════════════════════════════════════════════════════ section("3. AUTONOMOUS ENGINE API") s, body = api_get("/autonomous/status") if s == 200 and isinstance(body, dict): check("Engine status reachable", True, f"HTTP {s}") check("Running field present", "running" in body or "is_running" in body, str(list(body.keys()))[:80]) check("Mode field present", any(k in body for k in ("mode","trading_mode","paper")), str(list(body.keys()))[:80]) check("Cycle count present", any(k in body for k in ("cycle","cycles","cycle_count")), str(list(body.keys()))[:80]) else: check("Autonomous status", False, f"HTTP {s}") s, body = api_get("/autonomous/scan-results") if s == 200: check("Scan results reachable", True, f"HTTP {s}") if isinstance(body, list): check("Scan results is list", True, f"{len(body)} candidates") elif isinstance(body, dict): count = body.get("count", body.get("total", "?")) check("Scan results OK", True, f"count={count}") else: check("Scan results", False, f"HTTP {s}") s, body = api_get("/autonomous/capital-pools") if s == 200: check("Capital pools reachable", True) if isinstance(body, dict): pools = list(body.keys()) check("Pool names present", len(pools) > 0, f"pools={pools[:3]}") else: check("Capital pools", False, f"HTTP {s}") # ═══════════════════════════════════════════════════════════════ # SECTION 4: RISK GUARD API # ═══════════════════════════════════════════════════════════════ section("4. RISK GUARD API") s, body = api_get("/risk/status") if s == 200 and isinstance(body, dict): check("Risk Guard reachable", True) check("Kill switch field", any("kill" in k.lower() for k in body), str(list(body.keys()))[:80]) check("Loss limit field", any("loss" in k.lower() for k in body), str(list(body.keys()))[:80]) else: check("Risk Guard status", s == 200, f"HTTP {s}") # ═══════════════════════════════════════════════════════════════ # SECTION 5: F&O ENGINE API # ═══════════════════════════════════════════════════════════════ section("5. F&O ENGINE API") s, body = api_get("/fno/status") check("F&O Engine reachable", s == 200, f"HTTP {s}") s, body = api_get("/fno/lot-sizes") check("F&O lot sizes endpoint", s in (200, 404), f"HTTP {s}", warn=(s==404)) s, body = api_get("/fno/oi-analysis") check("F&O OI analysis endpoint", s in (200, 404), f"HTTP {s}", warn=(s==404)) # ═══════════════════════════════════════════════════════════════ # SECTION 6: LEARNING ENGINE API # ═══════════════════════════════════════════════════════════════ section("6. LEARNING ENGINE API") s, body = api_get("/learning/stats") if s == 200 and isinstance(body, dict): check("Learning stats reachable", True) check("Trade count field", any("trade" in k.lower() for k in body), str(list(body.keys()))[:80]) check("Win rate field", any("win" in k.lower() for k in body), str(list(body.keys()))[:80]) else: check("Learning Engine stats", s == 200, f"HTTP {s}") s, _ = api_get("/learning/suggestions") check("AI suggestions endpoint", s == 200, f"HTTP {s}") # ═══════════════════════════════════════════════════════════════ # SECTION 7: NEW MODULE FILES — EXISTENCE CHECK # ═══════════════════════════════════════════════════════════════ section("7. NEW ENHANCEMENT FILES — EXISTENCE") BASE = r"D:\CV6_AI_Trading_OS" new_files = [ (r"app\autonomous\smart_ranker.py", "SmartRanker — Phase 3 Ranking"), (r"app\autonomous\ai_context_builder.py","AIContextBuilder — Phase 4 Context"), (r"CV6_SCANNER_ENHANCEMENT_REPORT.md", "Validation Report"), (r"ARCHITECTURE_GAP_REPORT.md", "Architecture Gap Report"), (r"generate_cv6_ppt.py", "Educational PPT Generator"), ] modified_files = [ (r"app\autonomous\smart_scanner.py", "SmartScanner (wired ranker)"), (r"app\ai\ai_consensus.py", "AI Consensus (rich context)"), (r"app\autonomous\autonomous_engine.py", "Autonomous Engine (context builder)"), (r"app\risk\risk_router.py", "Risk Router (prefix fix)"), (r"app\fno\fno_engine.py", "FNO Engine"), (r"main.py", "Main FastAPI App"), ] p("\n New files:") for rel, label in new_files: path = os.path.join(BASE, rel) exists = os.path.isfile(path) size = os.path.getsize(path) if exists else 0 check(label, exists, f"{size:,} bytes" if exists else "MISSING") p("\n Modified files:") for rel, label in modified_files: path = os.path.join(BASE, rel) exists = os.path.isfile(path) size = os.path.getsize(path) if exists else 0 check(label, exists, f"{size:,} bytes" if exists else "MISSING") # ═══════════════════════════════════════════════════════════════ # SECTION 8: PYTHON SYNTAX CHECK — ALL KEY FILES # ═══════════════════════════════════════════════════════════════ section("8. PYTHON SYNTAX VALIDATION") py_files = [ r"app\autonomous\smart_ranker.py", r"app\autonomous\ai_context_builder.py", r"app\autonomous\smart_scanner.py", r"app\ai\ai_consensus.py", r"app\autonomous\autonomous_engine.py", r"app\risk\risk_router.py", r"app\fno\fno_engine.py", r"main.py", ] for rel in py_files: path = os.path.join(BASE, rel) if not os.path.isfile(path): check(rel, False, "FILE MISSING") continue try: with open(path, encoding="utf-8") as f: src = f.read() ast.parse(src) lines = src.count("\n") check(rel, True, f"{lines} lines — syntax OK") except SyntaxError as e: check(rel, False, f"SyntaxError line {e.lineno}: {e.msg}") except Exception as ex: check(rel, False, str(ex)[:60]) # ═══════════════════════════════════════════════════════════════ # SECTION 9: SMART RANKER — LOGIC TEST (no backend needed) # ═══════════════════════════════════════════════════════════════ section("9. SMART RANKER — UNIT TEST (offline)") try: sys.path.insert(0, BASE) from app.autonomous.smart_scanner import SmartCandidate, SmartScanner from app.autonomous.smart_ranker import SmartRanker, RankScore # Build mock candidates mock_candidates = [] for sym, score, sig in [ ("RELIANCE", 72.5, "BUY"), ("TCS", 68.0, "BUY"), ("INFY", 65.0, "SELL"), ]: c = SmartCandidate( symbol=sym, sector="IT" if sym in ("TCS","INFY") else "Energy", ltp=1000.0, change_pct=1.2, volume=1_000_000, tech_score=60, quality_score=65, risk_score=70, total_score=score, signal=sig, strategy="EMA_CROSSOVER", volume_ratio=2.0, rsi=55.0, ) mock_candidates.append(c) ranker = SmartRanker() ranked = ranker.rank(mock_candidates, top_n=3) check("SmartRanker instantiates", True) check("rank() returns list", isinstance(ranked, list), f"got {type(ranked)}") check("rank() respects top_n", len(ranked) <= 3, f"returned {len(ranked)}") check("rank_score attached", hasattr(ranked[0], "rank_score"), str(type(getattr(ranked[0],'rank_score',None)))) rs = ranked[0].rank_score check("RankScore.total in 0-100", 0 <= rs.total <= 100, f"{rs.total:.1f}") check("RankScore.composite in 0-100", 0 <= rs.composite <= 100, f"{rs.composite:.1f}") check("market_personality set", bool(rs.market_personality), rs.market_personality) check("mtf_agreement_pct in 0-100", 0 <= rs.mtf_agreement_pct <= 100, f"{rs.mtf_agreement_pct:.1f}%") check("Sorted by composite desc", all(ranked[i].rank_score.composite >= ranked[i+1].rank_score.composite for i in range(len(ranked)-1)), "descending order ✓") except Exception as ex: check("SmartRanker unit test", False, str(ex)[:80]) # ═══════════════════════════════════════════════════════════════ # SECTION 10: AI CONTEXT BUILDER — LOGIC TEST # ═══════════════════════════════════════════════════════════════ section("10. AI CONTEXT BUILDER — UNIT TEST (offline)") try: from app.autonomous.ai_context_builder import AIContextBuilder, AIContextPackage builder = AIContextBuilder() # Use a ranked candidate if available from section 9 test_cand = ranked[0] if ranked else mock_candidates[0] ctx = builder.build(test_cand, portfolio_state=None) check("AIContextBuilder instantiates", True) check("build() returns AIContextPackage", isinstance(ctx, AIContextPackage), str(type(ctx))) check("market section populated", bool(ctx.market.personality), ctx.market.personality) check("sector section populated", bool(ctx.sector.stock_sector), ctx.sector.stock_sector) check("liquidity section populated", ctx.liquidity.liquidity_score >= 0, f"{ctx.liquidity.liquidity_score:.1f}") check("MTF agreement % in 0-100", 0 <= ctx.mtf.agreement_pct <= 100, f"{ctx.mtf.agreement_pct:.1f}%") check("breadth signal set", bool(ctx.breadth.breadth_signal), ctx.breadth.breadth_signal) check("confidence.overall_confidence ≥ 0", ctx.confidence.overall_confidence >= 0, f"{ctx.confidence.overall_confidence:.1f}") prompt = ctx.to_prompt() check("to_prompt() returns non-empty string", bool(prompt) and len(prompt) > 100, f"{len(prompt)} chars") check("Prompt has SIGNAL instruction", "SIGNAL:" in prompt) check("Prompt has CONFIDENCE instruction", "CONFIDENCE:" in prompt) check("Prompt has RISK_LEVEL instruction", "RISK_LEVEL:" in prompt) check("Prompt has all 9 sections", all(sec in prompt for sec in [ "MARKET PERSONALITY","SECTOR ROTATION","LIQUIDITY", "INSTITUTIONAL","RELATIVE STRENGTH","MULTI-TIMEFRAME", "MARKET BREADTH","PORTFOLIO IMPACT","AI CONFIDENCE INPUTS" ]), "all 9 headers found") except Exception as ex: check("AIContextBuilder unit test", False, str(ex)[:80]) # ═══════════════════════════════════════════════════════════════ # SECTION 11: AI CONSENSUS — INTERFACE CHECK # ═══════════════════════════════════════════════════════════════ section("11. AI CONSENSUS ENGINE — INTERFACE CHECK") try: from app.ai.ai_consensus import AIConsensusEngine, ConsensusResult engine = AIConsensusEngine() check("AIConsensusEngine instantiates", True) check("context_package param in consensus()", "context_package" in engine.consensus.__code__.co_varnames, "parameter present ✓") check("_build_prompt accepts context_package", "context_package" in engine._build_prompt.__code__.co_varnames, "parameter present ✓") # Test rich prompt override rich_prompt = "TEST RICH PROMPT TEXT" result = engine._build_prompt("TESTSYM", {}, {"prompt_text": rich_prompt}) check("Rich context overrides basic prompt", result == rich_prompt, f"returned: '{result[:40]}'") # Test fallback when no context basic = engine._build_prompt("TESTSYM", {"ltp": 100, "change_pct": 1.5, "open":98,"high":102,"low":97, "close":100,"volume":500000}) check("Basic fallback prompt generated", "TESTSYM" in basic or "100" in basic, basic[:50]) except Exception as ex: check("AI Consensus interface check", False, str(ex)[:80]) # ═══════════════════════════════════════════════════════════════ # SECTION 12: FRONTEND CHECK # ═══════════════════════════════════════════════════════════════ section("12. FRONTEND (localhost:3000)") s, body = api_get("/", timeout=3) # this goes to backend # Check frontend separately try: req = urllib.request.Request("http://localhost:3000") with urllib.request.urlopen(req, timeout=3) as r: fe_status = r.status content = r.read(500).decode("utf-8", errors="ignore") check("Frontend running", fe_status == 200, f"HTTP {fe_status}") check("React app content", "CV6" in content or "react" in content.lower() or "