64 linhas
Sem EOL
2,2 KiB
Python
64 linhas
Sem EOL
2,2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Run the walk-forward experiment across all variants and write the research log.
|
|
|
|
Usage (from repo root):
|
|
python -m scripts.run_experiment --config configs/default.json --out results/oos_log.csv
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
|
|
from src.forecasting.target import ForecastContext
|
|
from src.forecasting.interface import Series
|
|
from src.arima import ArimaConfig
|
|
from src.sax import SaxConfig
|
|
from src.validation import WalkForwardConfig
|
|
from src.pipeline import build_models, run_variants, write_log
|
|
from tests.helpers import make_series
|
|
|
|
|
|
def load_config(path: str) -> dict:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def config_hash(cfg: dict) -> str:
|
|
return hashlib.sha256(json.dumps(cfg, sort_keys=True).encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", default="configs/default.json")
|
|
ap.add_argument("--out", default=None)
|
|
ap.add_argument("--bars", type=int, default=400)
|
|
args = ap.parse_args()
|
|
|
|
cfg = load_config(args.config)
|
|
h = cfg["horizon"]
|
|
ctx = ForecastContext(symbol=cfg["symbol"], timeframe=cfg["timeframe"],
|
|
horizon=h, atr_period=cfg["atr_period"])
|
|
arima_cfg = ArimaConfig(**cfg["arima"])
|
|
sax_cfg = SaxConfig(**cfg["sax"])
|
|
walk_cfg = WalkForwardConfig(
|
|
min_origin=cfg["data"]["train_length"],
|
|
n_forecast_points=cfg["data"]["train_length"] + 80,
|
|
step=cfg["data"]["step"],
|
|
)
|
|
|
|
series = make_series(n=args.bars, seed=cfg["reproducibility"]["seed"])
|
|
c_hash = config_hash(cfg)
|
|
|
|
variants = run_variants(series, ctx, arima_cfg, sax_cfg, walk_cfg,
|
|
configuration_hash=c_hash,
|
|
data_snapshot_id=f"synthetic-{args.bars}")
|
|
combined = list(variants["A_naive"]) + list(variants["B_arima"]) + list(variants["C_sax"])
|
|
out = args.out or os.path.join("results", "oos_log.csv")
|
|
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
|
write_log(combined, out)
|
|
print(f"Wrote {len(combined)} records -> {out}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |