Files
2026-06-26 20:50:07 +08:00

105 lines
3.7 KiB
Python

"""Interactive run-settings wizard (doc 05 §5).
Before a run, a small interactive wizard asks the settings that change
run-to-run and writes them to ``wizard-answers.yaml``. That YAML *is* the
reproducibility record: anyone can see exactly what produced an iteration's
numbers. Give every question a sensible default so a fast run is just
pressing Enter.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Optional
import yaml
@dataclass
class WizardQuestion:
"""One wizard question with a default and optional validator."""
key: str
prompt: str
default: Any
cast: Callable[[str], Any] = str
help: str = ""
def ask(self) -> Any:
"""Prompt the user, returning the cast value (default on empty)."""
suffix = f" [{self.help}]" if self.help else ""
raw = input(f"{self.prompt} (default: {self.default}){suffix}: ").strip()
if raw == "":
return self.default
try:
return self.cast(raw)
except (ValueError, TypeError):
print(f" invalid value, using default {self.default!r}")
return self.default
# Base set of common questions (doc 05 §5). Extend per-strategy via extra.
DEFAULT_QUESTIONS: list[WizardQuestion] = [
WizardQuestion("period_start", "Backtest start date (YYYY-MM-DD)", "2020-01-01"),
WizardQuestion("period_end", "Backtest end date (YYYY-MM-DD)", "2026-01-01"),
WizardQuestion("instrument_profile", "Instrument profile (real/worst_case/best_case)", "real"),
WizardQuestion("trials", "Optuna trial budget", 300, cast=int),
WizardQuestion("max_dd_ccy", "Hard drawdown cap (account currency)", 3500.0, cast=float),
WizardQuestion("max_dd_pct", "Hard drawdown cap (fraction of deposit)", 0.35, cast=float),
WizardQuestion("top_n_verify", "Finalists to send to MT5", 3, cast=int),
WizardQuestion("initial_deposit", "Initial deposit", 10000.0, cast=float),
WizardQuestion("n_jobs", "Optuna parallel workers", 4, cast=int),
]
@dataclass
class WizardAnswers:
"""A bag of answered wizard questions, serializable to YAML."""
answers: dict[str, Any] = field(default_factory=dict)
def get(self, key: str, default: Any = None) -> Any:
return self.answers.get(key, default)
def to_dict(self) -> dict[str, Any]:
return dict(self.answers)
def run_wizard(
questions: Optional[list[WizardQuestion]] = None,
*,
extra: Optional[list[WizardQuestion]] = None,
) -> WizardAnswers:
"""Run the interactive wizard and return :class:`WizardAnswers`.
``questions`` defaults to :data:`DEFAULT_QUESTIONS`; ``extra`` lets a
strategy add its own (e.g. a grid wizard adds depth/multiplier defaults).
"""
qs = list(questions or DEFAULT_QUESTIONS)
if extra:
qs.extend(extra)
answers: dict[str, Any] = {}
print("=== Pre-run wizard ===")
for q in qs:
answers[q.key] = q.ask()
print("=== Wizard complete ===")
return WizardAnswers(answers=answers)
def save_answers(answers: WizardAnswers, path: str | Path) -> None:
"""Write answers to a YAML file next to the iteration."""
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("w", encoding="utf-8") as f:
yaml.safe_dump(answers.to_dict(), f, sort_keys=False, allow_unicode=True)
def load_answers(path: str | Path) -> WizardAnswers:
"""Load answers from a previously-written YAML (for re-runs)."""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"wizard answers not found: {p}")
with p.open("r", encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
return WizardAnswers(answers=dict(data))