438 라인
EOL 없음
20 KiB
Python
438 라인
EOL 없음
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Frozen E1-E8 pre-registered research execution.
|
|
|
|
Runs the validated harness against the FROZEN bootstrap protocol
|
|
(branch main, pre-experiment commit 3fbd268, config hash e2370884...).
|
|
|
|
ZERO optimization. All primary comparisons are PAIRED BY ORIGIN.
|
|
No lookahead. Chronological walk-forward. Frozen cost assumptions.
|
|
|
|
Artifacts written under results/:
|
|
e1_naive e2_arima_vs_naive e3_sax_vs_naive e4_hybrid_vs_arima
|
|
e5_hybrid_vs_sax e6_agreement e7_disagreement e8_economic_robustness
|
|
E1-E8_MANIFEST.json E1-E8_RESULTS.csv
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from src.forecasting.target import ForecastContext, data_atr
|
|
from src.arima import ArimaConfig
|
|
from src.sax import SaxConfig
|
|
from src.validation import WalkForwardConfig
|
|
from src.pipeline import run_variants, build_hybrid_and_filters
|
|
from src.evaluation import metrics as M
|
|
from src.forecasting.record import HybridState
|
|
from src.data import validate_series
|
|
from tests.helpers import make_series
|
|
|
|
PROTOCOL_VERSION = "e1-e8-v0.1.0-frozen"
|
|
|
|
|
|
def load_config(path):
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def config_hash(cfg):
|
|
return hashlib.sha256(json.dumps(cfg, sort_keys=True).encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def git_head():
|
|
try:
|
|
return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
def build_dataset(cfg):
|
|
"""Frozen single dataset: placeholder-ts-bars, seed 42, n=max_bars (3000)."""
|
|
n = cfg["data"]["max_bars"]
|
|
seed = cfg["reproducibility"]["seed"]
|
|
series = make_series(n=n, seed=seed)
|
|
rep = validate_series(list(series.timestamp), list(series.open),
|
|
list(series.high), list(series.low), list(series.close))
|
|
close = np.asarray(series.close, dtype=float)
|
|
dh = hashlib.sha256(close.astype("<f8").tobytes()).hexdigest()[:16]
|
|
meta = {
|
|
"source": cfg["source"], "symbol": cfg["symbol"], "timeframe": cfg["timeframe"],
|
|
"first_timestamp": str(series.timestamp[0]), "last_timestamp": str(series.timestamp[-1]),
|
|
"n_bars": n, "integrity_ok": rep.ok, "integrity_errors": rep.errors,
|
|
"dataset_hash": dh, "seed": seed,
|
|
"data_note": "SYNTHETIC placeholder series (random-walk construction); "
|
|
"no exploitable structure by design.",
|
|
}
|
|
return series, meta
|
|
|
|
|
|
def rec_frame(records):
|
|
return pd.DataFrame([r.to_dict() for r in records])
|
|
|
|
|
|
def bootstrap_ci(diffs, B=2000, seed=42):
|
|
"""Mean and percentile bootstrap CI of paired differences."""
|
|
diffs = np.asarray(diffs, dtype=float)
|
|
diffs = diffs[np.isfinite(diffs)]
|
|
if diffs.size == 0:
|
|
return None
|
|
rng = np.random.default_rng(seed)
|
|
means = np.empty(B)
|
|
for i in range(B):
|
|
means[i] = np.mean(rng.choice(diffs, size=diffs.size, replace=True))
|
|
return (float(np.mean(diffs)), float(np.percentile(means, 2.5)),
|
|
float(np.percentile(means, 97.5)))
|
|
|
|
|
|
def build_study(records, origin_atr_ratio, rt_bps, mult=1.0):
|
|
"""Attach economic study columns to a record frame (aligned by origin)."""
|
|
df = rec_frame(records)
|
|
df["y_true"] = pd.to_numeric(df["actual_forward_return_ATR"], errors="coerce")
|
|
df["predicted"] = df["normalized_expected_return"]
|
|
df["forecast_direction"] = df["forecast_direction"].fillna("NEUTRAL")
|
|
df["direction"] = np.where(df["forecast_direction"] == "LONG", 1.0,
|
|
np.where(df["forecast_direction"] == "SHORT", -1.0, 0.0))
|
|
df["trial"] = df["forecast_direction"].isin(["LONG", "SHORT"])
|
|
ratio = np.asarray([origin_atr_ratio[int(o)] for o in df["forecast_origin"]])
|
|
df["cost_atr"] = np.where(df["trial"], (rt_bps / 10000.0) * ratio * mult, 0.0)
|
|
df["gross_atr"] = df["direction"] * df["y_true"]
|
|
df["net_atr"] = df["gross_atr"] - df["cost_atr"]
|
|
return df
|
|
|
|
|
|
def forecast_stats(df):
|
|
"""Point-forecast metrics over valid (pred & actual) paired origins."""
|
|
d = df[(df["predicted"].notna()) & (df["y_true"].notna())]
|
|
if d.empty:
|
|
return {"n": 0}
|
|
y = d["y_true"].to_numpy()
|
|
p = d["predicted"].to_numpy()
|
|
return {"n": int(len(d)), "mae": M.mae(y, p), "rmse": M.rmse(y, p),
|
|
"bias": float(np.mean(p - y))}
|
|
|
|
|
|
def dir_acc(df):
|
|
d = df[(df["trial"]) & (df["y_true"].notna())]
|
|
if d.empty:
|
|
return np.nan
|
|
hits = ((d["direction"] > 0) & (d["y_true"] > 0)) | ((d["direction"] < 0) & (d["y_true"] < 0))
|
|
return float(hits.mean())
|
|
|
|
|
|
def trade_stats(df):
|
|
"""Economic stats over directional (traded) forecasts."""
|
|
t = df[df["trial"]]
|
|
if t.empty:
|
|
return {"n_trades": 0, "coverage": 0.0, "gross_exp": np.nan, "net_exp": np.nan,
|
|
"gross_sum": 0.0, "net_sum": 0.0, "profit_factor": np.nan,
|
|
"max_drawdown": np.nan, "sharpe_per_trade": np.nan,
|
|
"dir_acc": np.nan, "mean_abs_net": np.nan}
|
|
net = t["net_atr"].to_numpy()
|
|
g = t["gross_atr"].to_numpy()
|
|
return {
|
|
"n_trades": int(len(t)), "coverage": float(len(t) / len(df)),
|
|
"gross_exp": float(np.mean(g)), "net_exp": float(np.mean(net)),
|
|
"gross_sum": float(np.sum(g)), "net_sum": float(np.sum(net)),
|
|
"profit_factor": M.profit_factor(net.tolist()),
|
|
"max_drawdown": M.max_drawdown(net.tolist()),
|
|
"sharpe_per_trade": M.sharpe(net.tolist(), scale=1.0),
|
|
"dir_acc": dir_acc(df),
|
|
"mean_abs_net": float(np.mean(np.abs(net))),
|
|
}
|
|
|
|
|
|
def write_json(obj, path):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(obj, fh, indent=2, default=str)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", default="configs/default.json")
|
|
ap.add_argument("--out", default="results")
|
|
ap.add_argument("--bars", type=int, default=0, help="override dataset bars (0=frozen max_bars)")
|
|
args = ap.parse_args()
|
|
|
|
cfg = load_config(args.config)
|
|
if args.bars:
|
|
cfg["data"]["max_bars"] = args.bars
|
|
c_hash = config_hash(cfg)
|
|
commit = git_head()
|
|
stamp = datetime.now(timezone.utc).isoformat()
|
|
|
|
series, dmeta = build_dataset(cfg)
|
|
ctx = ForecastContext(symbol=cfg["symbol"], timeframe=cfg["timeframe"],
|
|
horizon=cfg["horizon"], 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"])
|
|
|
|
variants = run_variants(series, ctx, arima_cfg, sax_cfg, walk_cfg,
|
|
configuration_hash=c_hash,
|
|
data_snapshot_id=dmeta["dataset_hash"])
|
|
arima_recs, sax_recs = variants["B_arima"], variants["C_sax"]
|
|
hybrid_set = build_hybrid_and_filters(arima_recs, sax_recs)
|
|
|
|
close = series.closes()
|
|
atr = data_atr(close, ctx.atr_period)
|
|
origin_atr_ratio = close / atr
|
|
|
|
comm_bps = cfg["costs"]["commission_per_trade"] * 10000.0
|
|
spread_bps = cfg["costs"]["spread_bps"]
|
|
rt_bps = 2.0 * comm_bps + spread_bps
|
|
cost_note = (f"roundtrip cost = 2*commission({comm_bps:.2f}bps) + spread({spread_bps:.2f}bps)"
|
|
f" = {rt_bps:.2f}bps; converted to ATR units via close/ATR at entry.")
|
|
|
|
chains = {
|
|
"A_naive": variants["A_naive"],
|
|
"A_naive_drift": variants["A_naive_drift"],
|
|
"B_arima": variants["B_arima"],
|
|
"C_sax": variants["C_sax"],
|
|
"D_hybrid": hybrid_set["D_hybrid"],
|
|
"E_agreement": hybrid_set["E_agreement"],
|
|
"F_rejection": hybrid_set["F_rejection"],
|
|
}
|
|
studies = {k: build_study(v, origin_atr_ratio, rt_bps) for k, v in chains.items()}
|
|
|
|
manifest = {
|
|
"experiment_id": "E1-E8", "protocol_version": PROTOCOL_VERSION,
|
|
"repository": "ARIMA_SAX_Hybrid_Forecaster", "branch": "main",
|
|
"commit_sha": commit, "pre_experiment_commit_sha": "3fbd268",
|
|
"executed_at": stamp, "config_hash": c_hash,
|
|
"dataset": dmeta,
|
|
"model_config": {
|
|
"arima": cfg["arima"], "sax": cfg["sax"], "hybrid": cfg["hybrid"],
|
|
"baselines": cfg["baselines"],
|
|
},
|
|
"evaluation": {
|
|
"target_definition": ctx.target_definition, "horizon": ctx.horizon,
|
|
"atr_period": ctx.atr_period, "walk_forward": cfg["data"],
|
|
"cost_interpretation": cost_note, "cost_multiplier": 1.0,
|
|
"multiple_testing_correction": "NONE (exploratory with respect to multiple hypotheses; documented)",
|
|
"bootstrap": {"B": 2000, "seed": cfg["reproducibility"]["seed"]},
|
|
},
|
|
}
|
|
out = args.out
|
|
os.makedirs(out, exist_ok=True)
|
|
for d in ["e1_naive", "e2_arima_vs_naive", "e3_sax_vs_naive", "e4_hybrid_vs_arima",
|
|
"e5_hybrid_vs_sax", "e6_agreement", "e7_disagreement", "e8_economic_robustness"]:
|
|
os.makedirs(os.path.join(out, d), exist_ok=True)
|
|
write_json(manifest, os.path.join(out, "E1-E8_MANIFEST.json"))
|
|
|
|
results_summary = {}
|
|
|
|
# ---------------- E1: NAIVE ----------------
|
|
e1 = {"forecast": forecast_stats(studies["A_naive"]),
|
|
"trade": trade_stats(studies["A_naive"]),
|
|
"interpretation": "Reference point for incremental predictive information. "
|
|
"Naive never trades (net P&L = 0 by construction)."}
|
|
write_json(e1, os.path.join(out, "e1_naive", "E1_NAIVE.json"))
|
|
results_summary["E1"] = {"naive_mae": e1["forecast"].get("mae"),
|
|
"naive_rmse": e1["forecast"].get("rmse"),
|
|
"n": e1["forecast"].get("n")}
|
|
|
|
# ---------------- E2: ARIMA vs NAIVE ----------------
|
|
nv = studies["A_naive"].copy()
|
|
ar = studies["B_arima"].copy()
|
|
j = nv[["forecast_origin", "y_true"]].merge(
|
|
ar[["forecast_origin", "predicted", "trial", "direction", "y_true"]],
|
|
on="forecast_origin", suffixes=("_naive", "_arima"))
|
|
j = j[(j["y_true_naive"].notna()) & (j["predicted"].notna())]
|
|
y = j["y_true_naive"].to_numpy()
|
|
p_ar = j["predicted"].to_numpy()
|
|
diff_mae = np.abs(y) - np.abs(y - p_ar)
|
|
ci = bootstrap_ci(diff_mae)
|
|
mae_n, mae_a = M.mae(y, np.zeros_like(y)), M.mae(y, p_ar)
|
|
e2 = {
|
|
"n_paired": int(len(j)),
|
|
"naive": {"mae": mae_n, "rmse": M.rmse(y, np.zeros_like(y))},
|
|
"arima": {"mae": mae_a, "rmse": M.rmse(y, p_ar),
|
|
"mase": (mae_a / mae_n) if mae_n else np.nan,
|
|
"bias": float(np.mean(p_ar - y)),
|
|
"dir_acc": dir_acc(ar), "trades": trade_stats(ar)},
|
|
"paired_mae_diff_naive_minus_arima": {"mean": ci[0], "ci95_low": ci[1], "ci95_high": ci[2]}
|
|
if ci else None,
|
|
"interpretation": "Positive paired MAE difference = ARIMA has lower absolute error.",
|
|
}
|
|
write_json(e2, os.path.join(out, "e2_arima_vs_naive", "E2_ARIMA_VS_NAIVE.json"))
|
|
results_summary["E2"] = {"arima_mae": mae_a, "naive_mae": mae_n,
|
|
"arima_mase": e2["arima"]["mase"],
|
|
"arima_dir_acc": e2["arima"]["dir_acc"],
|
|
"paired_mae_diff_mean": e2["paired_mae_diff_naive_minus_arima"]["mean"]}
|
|
|
|
# ---------------- E3: SAX vs NAIVE ----------------
|
|
sx = studies["C_sax"].copy()
|
|
j3 = nv[["forecast_origin", "y_true"]].merge(
|
|
sx[["forecast_origin", "predicted", "trial", "direction", "y_true", "sax_analog_count"]],
|
|
on="forecast_origin", suffixes=("_naive", "_sax"))
|
|
j3 = j3[(j3["y_true_naive"].notna()) & (j3["predicted"].notna())]
|
|
y3 = j3["y_true_naive"].to_numpy()
|
|
p_sx = j3["predicted"].to_numpy()
|
|
diff_mae3 = np.abs(y3) - np.abs(y3 - p_sx)
|
|
ci3 = bootstrap_ci(diff_mae3)
|
|
mae_n3, mae_s = M.mae(y3, np.zeros_like(y3)), M.mae(y3, p_sx)
|
|
e3 = {
|
|
"n_paired": int(len(j3)),
|
|
"sax_analog_count_stats": {
|
|
"mean": float(j3["sax_analog_count"].mean()),
|
|
"median": float(j3["sax_analog_count"].median()),
|
|
"min": int(j3["sax_analog_count"].min()),
|
|
"max": int(j3["sax_analog_count"].max()),
|
|
},
|
|
"naive": {"mae": mae_n3, "rmse": M.rmse(y3, np.zeros_like(y3))},
|
|
"sax": {"mae": mae_s, "rmse": M.rmse(y3, p_sx),
|
|
"mase": (mae_s / mae_n3) if mae_n3 else np.nan,
|
|
"bias": float(np.mean(p_sx - y3)),
|
|
"dir_acc": dir_acc(sx), "trades": trade_stats(sx)},
|
|
"paired_mae_diff_naive_minus_sax": {"mean": ci3[0], "ci95_low": ci3[1], "ci95_high": ci3[2]}
|
|
if ci3 else None,
|
|
"interpretation": "Positive paired MAE difference = SAX has lower absolute error.",
|
|
}
|
|
write_json(e3, os.path.join(out, "e3_sax_vs_naive", "E3_SAX_VS_NAIVE.json"))
|
|
results_summary["E3"] = {"sax_mae": mae_s, "sax_mase": e3["sax"]["mase"],
|
|
"sax_dir_acc": e3["sax"]["dir_acc"],
|
|
"paired_mae_diff_mean": e3["paired_mae_diff_naive_minus_sax"]["mean"]}
|
|
|
|
# ---------------- E4: HYBRID vs ARIMA ----------------
|
|
hy = studies["D_hybrid"].copy()
|
|
j4 = ar[["forecast_origin", "direction", "net_atr", "y_true"]].merge(
|
|
hy[["forecast_origin", "direction", "net_atr", "hybrid_state"]],
|
|
on="forecast_origin", suffixes=("_arima", "_hybrid"))
|
|
both = j4[(j4["direction_arima"] != 0) & (j4["direction_hybrid"] != 0)]
|
|
diff_net4 = both["net_atr_hybrid"] - both["net_atr_arima"]
|
|
ci4 = bootstrap_ci(diff_net4.to_numpy())
|
|
e4 = {
|
|
"n_paired_origins": int(len(j4)),
|
|
"n_trade_both": int(len(both)),
|
|
"arima_trades": trade_stats(ar), "hybrid_trades": trade_stats(hy),
|
|
"paired_net_diff_hybrid_minus_arima": {"mean": ci4[0], "ci95_low": ci4[1],
|
|
"ci95_high": ci4[2]} if ci4 else None,
|
|
"interpretation": "Positive paired net diff = hybrid net return exceeds ARIMA "
|
|
"at the same forecast origins (frozen cost).",
|
|
}
|
|
write_json(e4, os.path.join(out, "e4_hybrid_vs_arima", "E4_HYBRID_VS_ARIMA.json"))
|
|
results_summary["E4"] = {"n_trade_both": int(len(both)),
|
|
"paired_net_diff_mean": e4["paired_net_diff_hybrid_minus_arima"]["mean"]}
|
|
|
|
# ---------------- E5: HYBRID vs SAX ----------------
|
|
j5 = sx[["forecast_origin", "direction", "net_atr", "y_true"]].merge(
|
|
hy[["forecast_origin", "direction", "net_atr", "hybrid_state"]],
|
|
on="forecast_origin", suffixes=("_sax", "_hybrid"))
|
|
both5 = j5[(j5["direction_sax"] != 0) & (j5["direction_hybrid"] != 0)]
|
|
diff_net5 = both5["net_atr_hybrid"] - both5["net_atr_sax"]
|
|
ci5 = bootstrap_ci(diff_net5.to_numpy())
|
|
e5 = {
|
|
"n_paired_origins": int(len(j5)),
|
|
"n_trade_both": int(len(both5)),
|
|
"sax_trades": trade_stats(sx), "hybrid_trades": trade_stats(hy),
|
|
"paired_net_diff_hybrid_minus_sax": {"mean": ci5[0], "ci95_low": ci5[1],
|
|
"ci95_high": ci5[2]} if ci5 else None,
|
|
"interpretation": "Positive paired net diff = hybrid net return exceeds SAX "
|
|
"at the same forecast origins (frozen cost).",
|
|
}
|
|
write_json(e5, os.path.join(out, "e5_hybrid_vs_sax", "E5_HYBRID_VS_SAX.json"))
|
|
results_summary["E5"] = {"n_trade_both": int(len(both5)),
|
|
"paired_net_diff_mean": e5["paired_net_diff_hybrid_minus_sax"]["mean"]}
|
|
|
|
# ---------------- E6: AGREEMENT ANALYSIS ----------------
|
|
e6_rows = []
|
|
for state, g in hy.groupby("hybrid_state"):
|
|
g = g.copy()
|
|
row = {"hybrid_state": state, "n": int(len(g)),
|
|
"mean_y": float(g["y_true"].mean()),
|
|
"median_y": float(g["y_true"].median()),
|
|
"std_y": float(g["y_true"].std()),
|
|
"dir_acc": dir_acc(g), "net_exp": trade_stats(g)["net_exp"],
|
|
"coverage": trade_stats(g)["coverage"]}
|
|
e6_rows.append(row)
|
|
e6 = {"groups": e6_rows,
|
|
"interpretation": "Tests whether agreement (strong_agreement) has conditional "
|
|
"predictive value vs disagreement / partial / no-edge."}
|
|
write_json(e6, os.path.join(out, "e6_agreement", "E6_AGREEMENT.json"))
|
|
pd.DataFrame(e6_rows).to_csv(os.path.join(out, "e6_agreement", "E6_AGREEMENT.csv"), index=False)
|
|
results_summary["E6"] = {r["hybrid_state"]: {"n": r["n"], "mean_y": r["mean_y"],
|
|
"dir_acc": r["dir_acc"]} for r in e6_rows}
|
|
|
|
# ---------------- E7: DISAGREEMENT / REJECTION ----------------
|
|
acc = hy[hy["hybrid_state"].isin([HybridState.STRONG_AGREEMENT.value,
|
|
HybridState.PARTIAL_EVIDENCE.value,
|
|
HybridState.NO_EDGE.value])]
|
|
rej = hy[hy["hybrid_state"].isin([HybridState.DISAGREEMENT.value,
|
|
HybridState.INSUFFICIENT_EVIDENCE.value])]
|
|
e7 = {
|
|
"all": trade_stats(hy),
|
|
"accepted_F": trade_stats(studies["F_rejection"]),
|
|
"rejected": {"n": int(len(rej)), "mean_y": float(rej["y_true"].mean()),
|
|
"dir_acc": dir_acc(rej), "trade_stats": trade_stats(rej)},
|
|
"accepted_n": int(len(acc)),
|
|
"interpretation": "If accepted conditional outcomes are better than rejected, "
|
|
"disagreement/insufficiency is a useful (descriptive) rejection filter.",
|
|
}
|
|
write_json(e7, os.path.join(out, "e7_disagreement", "E7_REJECTION.json"))
|
|
results_summary["E7"] = {"all_net_exp": e7["all"]["net_exp"],
|
|
"accepted_net_exp": e7["accepted_F"]["net_exp"],
|
|
"rejected_net_exp": e7["rejected"]["trade_stats"]["net_exp"]}
|
|
|
|
# ---------------- E8: ECONOMIC + ROBUSTNESS ----------------
|
|
e8_tbl = {}
|
|
for k, s in studies.items():
|
|
st = trade_stats(s)
|
|
e8_tbl[k] = {"n": int(len(s)), "trades": st["n_trades"], "coverage": st["coverage"],
|
|
"gross_exp": st["gross_exp"], "net_exp": st["net_exp"],
|
|
"gross_sum": st["gross_sum"], "net_sum": st["net_sum"],
|
|
"profit_factor": st["profit_factor"], "max_drawdown": st["max_drawdown"],
|
|
"sharpe_per_trade": st["sharpe_per_trade"], "dir_acc": st["dir_acc"]}
|
|
cost_sens = {}
|
|
for mult in [0.0, 0.5, 1.0, 2.0, 5.0]:
|
|
cost_sens[str(mult)] = {k: trade_stats(build_study(v, origin_atr_ratio, rt_bps, mult))["net_exp"]
|
|
for k, v in chains.items()}
|
|
seg_rows = []
|
|
for k, s in studies.items():
|
|
s = s.sort_values("forecast_origin").reset_index(drop=True)
|
|
parts = np.array_split(np.arange(len(s)), 3)
|
|
for i, idx in enumerate(parts):
|
|
g = s.iloc[idx]
|
|
seg_rows.append({"variant": k, "segment": i + 1,
|
|
"origin_first": int(g["forecast_origin"].min()),
|
|
"origin_last": int(g["forecast_origin"].max()),
|
|
"n": int(len(g)), "mean_y": float(g["y_true"].mean()),
|
|
"dir_acc": dir_acc(g), "net_exp": trade_stats(g)["net_exp"]})
|
|
e8 = {"economic_table": e8_tbl, "cost_sensitivity_net_exp": cost_sens,
|
|
"segments": seg_rows,
|
|
"interpretation": "Robustness requires consistency across segments, not a single good period."}
|
|
write_json(e8, os.path.join(out, "e8_economic_robustness", "E8_ECONOMIC.json"))
|
|
pd.DataFrame(e8_tbl).T.to_csv(os.path.join(out, "e8_economic_robustness", "E8_ECONOMIC.csv"))
|
|
pd.DataFrame(cost_sens).to_csv(os.path.join(out, "e8_economic_robustness", "E8_COST_SENSITIVITY.csv"))
|
|
pd.DataFrame(seg_rows).to_csv(os.path.join(out, "e8_economic_robustness", "E8_SEGMENTS.csv"))
|
|
results_summary["E8"] = {"variants": e8_tbl, "cost_sensitivity": cost_sens,
|
|
"segments": seg_rows}
|
|
|
|
# ---------------- consolidated results ----------------
|
|
all_frames = []
|
|
for k, s in studies.items():
|
|
s = s.copy()
|
|
s = s.drop(columns=["variant"], errors="ignore")
|
|
s.insert(0, "chain", k)
|
|
all_frames.append(s)
|
|
full = pd.concat(all_frames, ignore_index=True)
|
|
full.to_csv(os.path.join(out, "E1-E8_RESULTS.csv"), index=False)
|
|
|
|
write_json(results_summary, os.path.join(out, "E1-E8_RESULTS_SUMMARY.json"))
|
|
print(f"E1-E8 complete. Manifest/artifacts under {out}")
|
|
print(f" origins per variant: {len(studies['A_naive'])}")
|
|
print(f" ARIMA trades: {trade_stats(studies['B_arima'])['n_trades']}, "
|
|
f"SAX trades: {trade_stats(studies['C_sax'])['n_trades']}, "
|
|
f"Hybrid trades: {trade_stats(studies['D_hybrid'])['n_trades']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |