Add trader mainline surface audit

2026-05-21 11:02:07 +08:00
parent db1edf5c66
commit fdd3e08ade
6 changed files with 614 additions and 4 deletions
@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+import argparse
+from collections import Counter, defaultdict
+import json
+from pathlib import Path
+import re
+
+
+ROOT = Path(__file__).resolve().parents[1]
+INDEX_HTML = ROOT / "src" / "web" / "index.html"
+APP_JS = ROOT / "src" / "web" / "app.js"
+TRADER_MODE_JS = ROOT / "src" / "web" / "trader_mode.js"
+API_SERVER = ROOT / "src" / "api" / "server.py"
+TESTS_DIR = ROOT / "tests" / "unit"
+
+MAINLINE_SECTION_IDS = {
+    "trader-mode",
+    "strategy-backtest",
+    "trading-plan",
+    "journal",
+}
+
+SYSTEM_OPS_SECTION_IDS = {
+    "system-ops",
+    "overview",
+    "trading-status",
+    "runtime-supervisor",
+    "decision-packet",
+    "readiness",
+    "session-focus",
+    "intraday-candidates",
+    "alerts",
+    "candles",
+}
+
+RESEARCH_ONLY_SECTION_IDS = {
+    "decision-assistant",
+    "operator-console",
+    "visual-reviews",
+}
+
+DEPRECATED_SECTION_IDS = {
+    "execution-tickets",
+    "position-management",
+}
+
+MAINLINE_ENDPOINT_PREFIXES = (
+    "/health",
+    "/candles",
+    "/trading/backtests",
+    "/trading/plans",
+    "/trading/review-packet",
+    "/trading/operator-notes",
+    "/trading/shadow-paper",
+    "/trading/shadow-paper-validation",
+    "/trading/journal",
+    "/trading/research/weekly-review",
+    "/trading/launch-readiness",
+)
+
+SYSTEM_OPS_ENDPOINT_PREFIXES = (
+    "/trading/broker-state",
+    "/trading/market-state",
+    "/trading/runtime",
+    "/trading/maintenance-summary",
+    "/trading/unattended-harness",
+    "/trading/trust-score",
+    "/trading/session-view",
+    "/trading/candidates",
+    "/trading/decision-packet",
+    "/trading/alerts",
+    "/trading/live-readiness",
+)
+
+RESEARCH_ONLY_ENDPOINT_PREFIXES = (
+    "/trading/archive-baselines",
+    "/trading/assistant-focus",
+    "/trading/external-context",
+    "/trading/research-council",
+    "/trading/visual-watch",
+    "/trading/workflow-archive",
+    "/training",
+)
+
+DEPRECATED_ENDPOINT_PREFIXES = (
+    "/live-operation-audits",
+    "/trading/okx-sandbox-preflight",
+    "/trading/reconcile",
+    "/trading/executions",
+    "/trading/tickets",
+)
+
+ROUTE_RE = re.compile(r'if method == "([A-Z]+)" and path == "([^"]+)"')
+SECTION_RE = re.compile(
+    r'data-workbench-section="([^"]+)"[^>]*data-workbench-label="([^"]+)"',
+    re.MULTILINE,
+)
+NAV_RE = re.compile(r'data-workbench-target="([^"]+)"[^>]*>([^<]+)</button>')
+SYSTEM_OPS_RE = re.compile(r'data-system-ops-target="([^"]+)"[^>]*>([^<]+)</button>')
+FRONTEND_ENDPOINT_RE = re.compile(
+    r"(?:fetchJson|postJson)\(\s*(?:`([^`]+)`|'([^']+)'|\"([^\"]+)\")"
+)
+TEST_ENDPOINT_RE = re.compile(
+    r"['\"](/(?:health|candles|trader-mode|static|live-operation-audits|training|trading)[^'\"]*)['\"]"
+)
+
+
+def normalize_endpoint(raw: str) -> str:
+    endpoint = raw.split("?", 1)[0].strip()
+    endpoint = re.sub(r"\$\{[^}]+\}", "{expr}", endpoint)
+    endpoint = re.sub(r"/\d+(?=/|$)", "/{id}", endpoint)
+    endpoint = endpoint.rstrip("/")
+    return endpoint or raw
+
+
+def classify_section(section_id: str) -> str:
+    if section_id in MAINLINE_SECTION_IDS:
+        return "Mainline"
+    if section_id in SYSTEM_OPS_SECTION_IDS:
+        return "System Ops"
+    if section_id in RESEARCH_ONLY_SECTION_IDS:
+        return "Research Only"
+    if section_id in DEPRECATED_SECTION_IDS:
+        return "Deprecated"
+    return "Unclassified"
+
+
+def classify_endpoint(endpoint: str) -> str:
+    normalized = normalize_endpoint(endpoint)
+    if normalized.startswith("/static") or normalized in {"/trader-mode", "/trader-mode.html"}:
+        return "Mainline"
+    if "/build-ticket" in normalized:
+        return "Deprecated"
+    if normalized.startswith(DEPRECATED_ENDPOINT_PREFIXES):
+        return "Deprecated"
+    if normalized.startswith(RESEARCH_ONLY_ENDPOINT_PREFIXES):
+        return "Research Only"
+    if normalized.startswith(MAINLINE_ENDPOINT_PREFIXES):
+        return "Mainline"
+    if normalized.startswith(SYSTEM_OPS_ENDPOINT_PREFIXES):
+        return "System Ops"
+    if normalized == "/trading/watch/run":
+        return "System Ops"
+    return "Unclassified"
+
+
+def read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8")
+
+
+def collect_workspace_sections(index_html: Path = INDEX_HTML) -> list[dict]:
+    text = read_text(index_html)
+    return [
+        {
+            "id": section_id,
+            "label": label,
+            "category": classify_section(section_id),
+        }
+        for section_id, label in SECTION_RE.findall(text)
+    ]
+
+
+def collect_nav_targets(index_html: Path = INDEX_HTML) -> list[dict]:
+    text = read_text(index_html)
+    return [
+        {
+            "id": target,
+            "label": label.strip(),
+            "category": classify_section(target),
+        }
+        for target, label in NAV_RE.findall(text)
+    ]
+
+
+def collect_system_ops_targets(index_html: Path = INDEX_HTML) -> list[dict]:
+    text = read_text(index_html)
+    return [
+        {
+            "id": target,
+            "label": label.strip(),
+            "category": classify_section(target),
+        }
+        for target, label in SYSTEM_OPS_RE.findall(text)
+    ]
+
+
+def collect_api_routes(api_server: Path = API_SERVER) -> list[dict]:
+    text = read_text(api_server)
+    routes = []
+    for method, endpoint in ROUTE_RE.findall(text):
+        normalized = normalize_endpoint(endpoint)
+        routes.append(
+            {
+                "method": method,
+                "endpoint": normalized,
+                "category": classify_endpoint(normalized),
+                "source": "src/api/server.py",
+            }
+        )
+    return sorted(routes, key=lambda item: (item["endpoint"], item["method"]))
+
+
+def collect_frontend_api_references(paths=None) -> list[dict]:
+    paths = [APP_JS, TRADER_MODE_JS] if paths is None else paths
+    references = {}
+    for path in paths:
+        text = read_text(path)
+        for match in FRONTEND_ENDPOINT_RE.findall(text):
+            raw = next(part for part in match if part)
+            endpoint = normalize_endpoint(raw)
+            references.setdefault(endpoint, set()).add(display_path(path))
+    return [
+        {
+            "endpoint": endpoint,
+            "category": classify_endpoint(endpoint),
+            "sources": sorted(sources),
+        }
+        for endpoint, sources in sorted(references.items())
+    ]
+
+
+def collect_test_endpoint_references(tests_dir: Path = TESTS_DIR) -> list[dict]:
+    references = defaultdict(set)
+    for path in sorted(tests_dir.glob("test_*.py")):
+        text = read_text(path)
+        for raw in TEST_ENDPOINT_RE.findall(text):
+            endpoint = normalize_endpoint(raw)
+            references[endpoint].add(display_path(path))
+    return [
+        {
+            "endpoint": endpoint,
+            "category": classify_endpoint(endpoint),
+            "sources": sorted(sources),
+        }
+        for endpoint, sources in sorted(references.items())
+    ]
+
+
+def build_audit() -> dict:
+    sections = collect_workspace_sections()
+    nav_targets = collect_nav_targets()
+    system_ops_targets = collect_system_ops_targets()
+    api_routes = collect_api_routes()
+    frontend_references = collect_frontend_api_references()
+    test_references = collect_test_endpoint_references()
+
+    endpoint_categories = Counter(item["category"] for item in api_routes)
+    frontend_categories = Counter(item["category"] for item in frontend_references)
+    section_categories = Counter(item["category"] for item in sections)
+
+    return {
+        "policy": {
+            "mainline_chain": (
+                "Range / Draw / Location -> Liquidity / Sweep / Inducement / Path "
+                "-> Displacement / MSS / FVG / Retest / Session -> backtest evidence "
+                "-> pre-market plan -> intraday focus/no-trade -> journal feedback"
+            ),
+            "execution_boundary": "read_only_analysis_no_order_submission",
+            "physical_delete_allowed": False,
+        },
+        "nav_targets": nav_targets,
+        "workspace_sections": sections,
+        "system_ops_targets": system_ops_targets,
+        "api_routes": api_routes,
+        "frontend_api_references": frontend_references,
+        "test_endpoint_references": test_references,
+        "summary": {
+            "nav_count": len(nav_targets),
+            "workspace_section_count": len(sections),
+            "system_ops_target_count": len(system_ops_targets),
+            "api_route_count": len(api_routes),
+            "frontend_api_reference_count": len(frontend_references),
+            "test_endpoint_reference_count": len(test_references),
+            "section_categories": dict(sorted(section_categories.items())),
+            "api_route_categories": dict(sorted(endpoint_categories.items())),
+            "frontend_reference_categories": dict(sorted(frontend_categories.items())),
+        },
+    }
+
+
+def format_markdown(audit: dict) -> str:
+    lines = [
+        "# Trader Mainline Surface Audit",
+        "",
+        f"- Execution boundary: `{audit['policy']['execution_boundary']}`",
+        f"- Physical delete allowed now: `{str(audit['policy']['physical_delete_allowed']).lower()}`",
+        f"- Mainline chain: {audit['policy']['mainline_chain']}",
+        "",
+        "## Navigation",
+        "",
+        "| Target | Label | Category |",
+        "| --- | --- | --- |",
+    ]
+    for item in audit["nav_targets"]:
+        lines.append(f"| `{item['id']}` | {item['label']} | {item['category']} |")
+    lines.extend(["", "## Workspace Sections", "", "| Section | Label | Category |", "| --- | --- | --- |"])
+    for item in audit["workspace_sections"]:
+        lines.append(f"| `{item['id']}` | {item['label']} | {item['category']} |")
+    lines.extend(["", "## API Route Category Counts", "", "```json"])
+    lines.append(json.dumps(audit["summary"]["api_route_categories"], ensure_ascii=False, sort_keys=True, indent=2))
+    lines.extend(["```", ""])
+    return "\n".join(lines)
+
+
+def display_path(path: Path) -> str:
+    try:
+        return str(path.relative_to(ROOT)).replace("\\", "/")
+    except ValueError:
+        return str(path)
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description="Audit trader-facing surfaces against the ICT mainline plan.")
+    parser.add_argument("--output-json", action="store_true", help="Print JSON instead of Markdown.")
+    args = parser.parse_args(argv)
+    audit = build_audit()
+    if args.output_json:
+        print(json.dumps(audit, ensure_ascii=False, sort_keys=True, indent=2))
+    else:
+        print(format_markdown(audit))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())