TradeBot/tools/accept_torch_candidate.py

from __future__ import annotations

import argparse
import json
import shutil
from pathlib import Path
from typing import Any


def main() -> None:
    args = _parse_args()
    current = _read_json(args.current_report)
    candidate = _read_json(args.candidate_report)
    decision = _decision(
        current,
        candidate,
        min_trades=args.min_trades,
        min_profit_factor=args.min_profit_factor,
        min_avg_net_percent=args.min_avg_net_percent,
        max_score_regression=args.max_score_regression,
    )
    payload = {
        "accepted": decision["accepted"],
        "reason": decision["reason"],
        "current": _summary(current),
        "candidate": _summary(candidate),
    }
    if args.report:
        Path(args.report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
    if not decision["accepted"]:
        raise SystemExit(2)
    target = Path(args.target_artifact)
    candidate_artifact = Path(args.candidate_artifact)
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(candidate_artifact, target)


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Accept or reject a retrained Torch candidate artifact.")
    parser.add_argument("--current-report", required=True)
    parser.add_argument("--candidate-report", required=True)
    parser.add_argument("--candidate-artifact", required=True)
    parser.add_argument("--target-artifact", required=True)
    parser.add_argument("--report", default="")
    parser.add_argument("--min-trades", type=int, default=8)
    parser.add_argument("--min-profit-factor", type=float, default=1.05)
    parser.add_argument("--min-avg-net-percent", type=float, default=0.0)
    parser.add_argument("--max-score-regression", type=float, default=0.05)
    return parser.parse_args()


def _decision(
    current: dict[str, Any],
    candidate: dict[str, Any],
    *,
    min_trades: int,
    min_profit_factor: float,
    min_avg_net_percent: float,
    max_score_regression: float,
) -> dict[str, Any]:
    candidate_score = _score(candidate)
    current_score = _score(current)
    candidate_replay = candidate.get("full_replay") if isinstance(candidate.get("full_replay"), dict) else {}
    candidate_walk = candidate.get("walk_forward") if isinstance(candidate.get("walk_forward"), dict) else {}
    walk_summary = candidate_walk.get("summary") if isinstance(candidate_walk.get("summary"), dict) else {}
    if not _validation_passed(candidate):
        return {"accepted": False, "reason": "candidate_failed_honest_validation"}
    if int(candidate_replay.get("trades", 0) or 0) < min_trades:
        return {"accepted": False, "reason": "candidate_has_too_few_full_replay_trades"}
    if float(candidate_replay.get("profit_factor", 0.0) or 0.0) < min_profit_factor:
        return {"accepted": False, "reason": "candidate_profit_factor_below_min"}
    if float(candidate_replay.get("avg_net_percent", 0.0) or 0.0) <= min_avg_net_percent:
        return {"accepted": False, "reason": "candidate_expectancy_non_positive"}
    if int(walk_summary.get("trades", 0) or 0) >= min_trades and float(walk_summary.get("avg_net_percent", 0.0) or 0.0) <= min_avg_net_percent:
        return {"accepted": False, "reason": "candidate_walk_forward_expectancy_non_positive"}
    if current_score > 0 and candidate_score < current_score * (1.0 - max_score_regression):
        return {"accepted": False, "reason": "candidate_score_regressed_vs_current"}
    return {"accepted": True, "reason": "candidate_passed_guard"}


def _validation_passed(report: dict[str, Any]) -> bool:
    validation = report.get("validation")
    if not isinstance(validation, dict):
        return False
    if "passed" in validation:
        return bool(validation.get("passed"))
    return str(validation.get("status", "")).strip().lower() in {"pass", "passed", "ok"}


def _score(report: dict[str, Any]) -> float:
    replay = report.get("full_replay") if isinstance(report.get("full_replay"), dict) else {}
    recommended = report.get("recommended") if isinstance(report.get("recommended"), dict) else {}
    replay_score = (
        float(replay.get("avg_net_percent", 0.0) or 0.0)
        + float(replay.get("total_net_percent", 0.0) or 0.0) * 0.02
        - float(replay.get("max_drawdown_percent", 0.0) or 0.0) * 0.05
        + min(float(replay.get("profit_factor", 0.0) or 0.0), 10.0) * 0.03
    )
    return replay_score + float(recommended.get("score", 0.0) or 0.0) * 0.25


def _summary(report: dict[str, Any]) -> dict[str, Any]:
    return {
        "score": round(_score(report), 6),
        "recommended": report.get("recommended", {}),
        "full_replay": report.get("full_replay", {}),
        "walk_forward_summary": (report.get("walk_forward") or {}).get("summary", {})
        if isinstance(report.get("walk_forward"), dict)
        else {},
        "benchmark_summary": (report.get("benchmark") or {}).get("summary", {})
        if isinstance(report.get("benchmark"), dict)
        else {},
        "validation": report.get("validation", {}),
    }


def _read_json(path: str) -> dict[str, Any]:
    if not path:
        return {}
    try:
        data = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


if __name__ == "__main__":
    main()