251 lines
7.4 KiB
Python
251 lines
7.4 KiB
Python
"""Run S/R detector, hidden-feature, or locked validation comparisons.
|
|
|
|
This is a cross-platform orchestrator around ``run_backtest_snapshot.py``. It
|
|
contains no backtest logic; every arm still runs through the production-parity
|
|
Python harness.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER = ROOT / "scripts" / "run_backtest_snapshot.py"
|
|
COMPARE = ROOT / "scripts" / "compare_sr_variants.py"
|
|
TRAINING_ARMS = (
|
|
"production_control",
|
|
"rewrite_legacy_primary",
|
|
"soft_zones_legacy_primary",
|
|
"confirmed_rounds_legacy_primary",
|
|
"gate_v2_legacy_primary",
|
|
)
|
|
RANGE_GRID_ARMS = (
|
|
"legacy_traffic_grid_only",
|
|
"legacy_range_grid_touch",
|
|
"legacy_range_grid_neutral",
|
|
)
|
|
LOCKABLE_ARMS = TRAINING_ARMS[1:] + RANGE_GRID_ARMS
|
|
TRAFFIC_ARMS = (
|
|
"production_control",
|
|
"legacy_geometry_neutral",
|
|
"legacy_pivots_only",
|
|
*RANGE_GRID_ARMS,
|
|
)
|
|
RANGE_FACTOR_ARMS = (
|
|
"production_range504",
|
|
"rewrite_range504_legacy_primary",
|
|
)
|
|
RANGE_RESIDUAL_ARMS = (
|
|
"rewrite_range504_structural_legacy_primary",
|
|
"rewrite_range504_structural_primary2",
|
|
)
|
|
FULL_PERIOD_ARMS = (
|
|
"production_control",
|
|
"rewrite_range504_structural_legacy_primary",
|
|
)
|
|
|
|
|
|
def _add_common(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument(
|
|
"--snapshot",
|
|
default="backtest_snapshots/prod.sqlite",
|
|
help="Local SQLite snapshot path.",
|
|
)
|
|
parser.add_argument("--workers", type=int, default=7)
|
|
|
|
|
|
def _args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
commands = parser.add_subparsers(dest="command", required=True)
|
|
train = commands.add_parser("train", help="Run all arms before 2024-07-01.")
|
|
_add_common(train)
|
|
traffic = commands.add_parser(
|
|
"traffic",
|
|
help="Isolate pivots, range-grid geometry, and touch strength on training data.",
|
|
)
|
|
_add_common(traffic)
|
|
traffic.add_argument(
|
|
"--only-arm",
|
|
choices=TRAFFIC_ARMS,
|
|
default=None,
|
|
help="Rerun one hidden-feature arm without repeating the full matrix.",
|
|
)
|
|
factor = commands.add_parser(
|
|
"factor",
|
|
help="Test the explicit 504-day range factor with old and clean detectors.",
|
|
)
|
|
_add_common(factor)
|
|
factor.add_argument(
|
|
"--only-arm",
|
|
choices=RANGE_FACTOR_ARMS,
|
|
default=None,
|
|
help="Run one range-factor arm without repeating the pair.",
|
|
)
|
|
residual = commands.add_parser(
|
|
"residual",
|
|
help="Isolate standalone rounds and the 1.5-2R primary-target veto.",
|
|
)
|
|
_add_common(residual)
|
|
residual.add_argument(
|
|
"--only-arm",
|
|
choices=RANGE_RESIDUAL_ARMS,
|
|
default=None,
|
|
help="Run one residual target-geometry arm without repeating the pair.",
|
|
)
|
|
full = commands.add_parser(
|
|
"full",
|
|
help="Run current production and the frozen candidate over the full snapshot.",
|
|
)
|
|
_add_common(full)
|
|
overlay = commands.add_parser(
|
|
"overlay",
|
|
help="Test the frozen 5% clean-structure rank overlay on production breadth.",
|
|
)
|
|
_add_common(overlay)
|
|
validate = commands.add_parser(
|
|
"validate",
|
|
help="Run production control and one locked arm from 2024-07-01.",
|
|
)
|
|
_add_common(validate)
|
|
validate.add_argument("--locked-arm", required=True, choices=LOCKABLE_ARMS)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _run_arm(
|
|
arm: str,
|
|
snapshot: str,
|
|
workers: int,
|
|
*,
|
|
entry_flag: str | None = None,
|
|
entry_date: str | None = None,
|
|
output: Path,
|
|
) -> None:
|
|
print(f"Running S/R arm: {arm}", flush=True)
|
|
command = [
|
|
sys.executable,
|
|
str(RUNNER),
|
|
snapshot,
|
|
"--workers", str(workers),
|
|
"--allow-spawn",
|
|
"--sr-variant", arm,
|
|
"--sr-audit",
|
|
"--out", str(output),
|
|
]
|
|
if entry_flag is not None and entry_date is not None:
|
|
command.extend((entry_flag, entry_date))
|
|
elif entry_flag is not None or entry_date is not None:
|
|
raise ValueError("entry_flag and entry_date must be provided together")
|
|
subprocess.run(
|
|
command,
|
|
cwd=ROOT,
|
|
check=True,
|
|
)
|
|
|
|
|
|
def _train(
|
|
args: argparse.Namespace,
|
|
arms: tuple[str, ...],
|
|
filename_prefix: str,
|
|
) -> None:
|
|
for arm in arms:
|
|
_run_arm(
|
|
arm,
|
|
args.snapshot,
|
|
args.workers,
|
|
entry_flag="--entry-end",
|
|
entry_date="2024-06-30",
|
|
output=ROOT / "reports" / f"{filename_prefix}-{arm}.json",
|
|
)
|
|
print("Training matrix complete. Review results before running validation.")
|
|
|
|
|
|
def _validate(args: argparse.Namespace) -> None:
|
|
reports: dict[str, Path] = {}
|
|
for arm in ("production_control", args.locked_arm):
|
|
output = ROOT / "reports" / f"backtest-sr-v2-validation-{arm}.json"
|
|
reports[arm] = output
|
|
_run_arm(
|
|
arm,
|
|
args.snapshot,
|
|
args.workers,
|
|
entry_flag="--entry-start",
|
|
entry_date="2024-07-01",
|
|
output=output,
|
|
)
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(COMPARE),
|
|
str(reports["production_control"]),
|
|
str(reports[args.locked_arm]),
|
|
"--out-csv", str(ROOT / "reports" / "sr-v2-validation-cohorts.csv"),
|
|
"--out-json", str(ROOT / "reports" / "sr-v2-validation-comparison.json"),
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
)
|
|
|
|
|
|
def _full(args: argparse.Namespace) -> None:
|
|
reports: dict[str, Path] = {}
|
|
for arm in FULL_PERIOD_ARMS:
|
|
output = ROOT / "reports" / f"backtest-sr-full-{arm}.json"
|
|
reports[arm] = output
|
|
_run_arm(
|
|
arm,
|
|
args.snapshot,
|
|
args.workers,
|
|
output=output,
|
|
)
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(COMPARE),
|
|
str(reports["production_control"]),
|
|
str(reports["rewrite_range504_structural_legacy_primary"]),
|
|
"--out-csv", str(ROOT / "reports" / "sr-full-production-vs-candidate-cohorts.csv"),
|
|
"--out-json", str(ROOT / "reports" / "sr-full-production-vs-candidate-comparison.json"),
|
|
],
|
|
cwd=ROOT,
|
|
check=True,
|
|
)
|
|
print("Full-period production comparison complete.")
|
|
|
|
|
|
def _overlay(args: argparse.Namespace) -> None:
|
|
_run_arm(
|
|
"production_structural_overlay",
|
|
args.snapshot,
|
|
args.workers,
|
|
output=ROOT / "reports" / "backtest-sr-overlay-full.json",
|
|
)
|
|
print("Full-period structural ranking overlay complete.")
|
|
|
|
|
|
def main() -> None:
|
|
args = _args()
|
|
if args.command == "train":
|
|
_train(args, TRAINING_ARMS, "backtest-sr-v2-train")
|
|
elif args.command == "traffic":
|
|
arms = (args.only_arm,) if args.only_arm else TRAFFIC_ARMS
|
|
_train(args, arms, "backtest-sr-traffic-train")
|
|
elif args.command == "factor":
|
|
arms = (args.only_arm,) if args.only_arm else RANGE_FACTOR_ARMS
|
|
_train(args, arms, "backtest-sr-range-factor-train")
|
|
elif args.command == "residual":
|
|
arms = (args.only_arm,) if args.only_arm else RANGE_RESIDUAL_ARMS
|
|
_train(args, arms, "backtest-sr-range-residual-train")
|
|
elif args.command == "full":
|
|
_full(args)
|
|
elif args.command == "overlay":
|
|
_overlay(args)
|
|
else:
|
|
_validate(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|