# -*- coding: utf-8 -*- """チャレンジ蒸留を「実データ・実戦略」で検証する。 これまでの当サイトの検証はコイン投げ（合成リターン）だった。 本スクリプトは 2010-07〜2026-09 の実価格から 5戦略 × 8銘柄 = 40本の「波」を作り、**実際の相関構造のまま**チャレンジを走らせる。 戦略は移動平均クロス / ボリンジャー逆張り / ドンチャンブレイク / RSI逆張り / モメンタム。 日足・シグナルは前日終値まで・約定は翌日（ルックアヘッドなし）・往復2bpのコスト込み。 各波は日次ボラを 0.8% に揃えてあるので、DD や目標の % が全波で同じ意味になる。 出力する3つの表: A … 口座数を固定し「別々の波の数」だけを変える（＝分散そのものの効果） B … 選び方（無作為 / 過去ベスト）で、選んだ波どうしの相関がどうなるか C … 相関 ρ のときリスクがどこまで下がるかの理論値 トライアルは実カレンダー上の開始日をランダム抽出する方式。 時系列の自己相関も、波どうしの同時相関も、実物がそのまま入る。 使い方: python scripts/fetch_wave_data.py # 先に価格データを取得 python scripts/sim-distillation-real-data.py """ import os import sys import numpy as np sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from lib_waves import build_waves # noqa: E402 SEED, TRIALS, DAYS = 20260918, 4000, 260 ACCOUNT, FEE, SPLIT = 100_000.0, 500.0, 0.80 P1, P2, MAX_DD, DAILY_DD = 0.08, 0.05, 0.10, 0.05 RANK_LOOKBACK = 500 # 「過去ベスト」を決める窓 def simulate(R): """R: (trials, days, n_acc) の日次リターン。口座ごとの受取額・到達・死亡を返す。""" T, D, N = R.shape eq = np.ones((T, N)) alive = np.ones((T, N), dtype=bool) stage = np.zeros((T, N), dtype=int) tgt = np.array([P1, P2, np.inf]) for d in range(D): step = R[:, d, :] eq = eq + step * alive # 日足なので「その日の損失」はこのステップそのもの。日次DDは毎日リセットされる。 alive &= ~(alive & (step <= -DAILY_DD)) alive &= ~(alive & (eq <= 1.0 - MAX_DD)) ok = alive & (stage < 2) & (eq >= 1.0 + tgt[np.minimum(stage, 2)]) if ok.any(): stage = np.where(ok, stage + 1, stage) eq = np.where(ok, 1.0, eq) funded = alive & (stage == 2) pay = np.where(funded, np.maximum(eq - 1.0, 0.0), 0.0) * ACCOUNT * SPLIT return pay, funded, ~alive def _stats(pay, funded, dead, n_acc): net = pay.sum(1) - FEE * n_acc return dict(全滅率=dead.all(1).mean(), 到達=funded.any(1).mean(), 本数=funded.sum(1).mean(), 手取り=net.mean(), プラス率=(net > 0).mean(), SD=net.std()) def _gather(Wv, starts, waves): R = np.empty((len(starts), DAYS, waves.shape[1])) for i, s in enumerate(starts): R[i] = Wv[s:s + DAYS][:, waves[i]] return R def run_distinct(Wv, n_acc, k, seed=SEED): """口座数 n_acc のうち、別々の波を k 本だけ使う。k=1 が完全な集中。""" rng = np.random.default_rng(seed) M = Wv.shape[1] starts = rng.integers(RANK_LOOKBACK, len(Wv) - DAYS, size=TRIALS) waves = np.zeros((TRIALS, n_acc), dtype=int) rhos = [] for i in range(TRIALS): sel = rng.choice(M, size=k, replace=False) waves[i] = np.resize(sel, n_acc) if k > 1: c = np.corrcoef(Wv[:, sel].T) rhos.append(c[np.triu_indices(k, 1)].mean()) pay, funded, dead = simulate(_gather(Wv, starts, waves)) out = _stats(pay, funded, dead, n_acc) out["ρ"] = float(np.mean(rhos)) if rhos else 1.0 return out def run_mode(Wv, mode, n_acc, seed=SEED): """集中/分散 × 無作為/過去ベスト の4通り。""" rng = np.random.default_rng(seed) M = Wv.shape[1] starts = rng.integers(RANK_LOOKBACK, len(Wv) - DAYS, size=TRIALS) if mode == "集中(無作為)": waves = np.repeat(rng.integers(0, M, size=(TRIALS, 1)), n_acc, axis=1) elif mode == "分散(無作為)": waves = np.array([rng.choice(M, size=n_acc, replace=False) for _ in range(TRIALS)]) else: waves = np.zeros((TRIALS, n_acc), dtype=int) for i, s in enumerate(starts): past = Wv[max(0, s - RANK_LOOKBACK):s] order = np.argsort(-(past.mean(0) / (past.std(0) + 1e-12))) waves[i] = order[:1].repeat(n_acc) if mode == "集中(過去ベスト)" else order[:n_acc] pay, funded, dead = simulate(_gather(Wv, starts, waves)) return _stats(pay, funded, dead, n_acc) def selection_rho(Wv, k, seed=SEED, n_samples=600): """無作為に k 本選んだときと、過去ベスト k 本のときの、波どうしの平均相関。""" rng = np.random.default_rng(seed) starts = rng.integers(RANK_LOOKBACK, len(Wv) - DAYS, size=n_samples) ixu = np.triu_indices(k, 1) rand, best = [], [] for s in starts: sel = rng.choice(Wv.shape[1], size=k, replace=False) rand.append(np.corrcoef(Wv[:, sel].T)[ixu].mean()) past = Wv[max(0, s - RANK_LOOKBACK):s] top = np.argsort(-(past.mean(0) / (past.std(0) + 1e-12)))[:k] best.append(np.corrcoef(Wv[:, top].T)[ixu].mean()) return float(np.mean(rand)), float(np.mean(best)) def risk_ratio(rho, n): """相関 ρ の波を N 本並べたときの、1本あたり比のリスク。下限は √ρ。""" return np.sqrt((1 + (n - 1) * rho) / n) if __name__ == "__main__": W = build_waves() Wv = W.values c = np.corrcoef(Wv.T) iu = np.triu_indices_from(c, 1) print(f"波 {Wv.shape[1]}本 / {W.index[0].date()}〜{W.index[-1].date()} / {len(W)}営業日") print(f"波どうしの平均相関 ρ={c[iu].mean():.3f}（最大 {c[iu].max():.2f} / 最小 {c[iu].min():.2f}）") print(f"$100K・P1 {P1:.0%}→P2 {P2:.0%}・最大DD {MAX_DD:.0%}・日次 {DAILY_DD:.0%}" f"・参加費 ${FEE:,.0f}・{DAYS}営業日・トライアル {TRIALS:,}") print("\n" + "=" * 104) print(" 【A】口座数10で固定し、『別々の波の数』だけ変える（総参加費は $5,000 で一定）") print("=" * 104) print(f" {'別々の波':12}{'全滅率':>9}{'1本以上到達':>13}{'期待手取り':>14}" f"{'プラス率':>10}{'標準偏差':>12}{'選んだ波のρ':>15}") for k in (1, 2, 3, 5, 10): r = run_distinct(Wv, 10, k) print(f" {k:>6}本 {r['全滅率']*100:8.1f}%{r['到達']*100:12.1f}%{r['手取り']:+14,.0f}" f"{r['プラス率']*100:9.1f}%{r['SD']:12,.0f}{r['ρ']:>14.3f}") print("\n" + "=" * 104) print(" 【A-2】口座数を増やしたとき、集中と分散で何が違うか") print("=" * 104) for n in (1, 5, 10, 20): print(f"\n 口座数 {n}（総参加費 ${FEE*n:,.0f}）") print(f" {'構成':22}{'全滅率':>9}{'1本以上到達':>13}{'期待手取り':>14}{'プラス率':>10}{'標準偏差':>12}") for mode in ("集中(無作為)", "集中(過去ベスト)", "分散(無作為)", "分散(過去ベスト)"): if n == 1 and mode.startswith("分散"): continue r = run_mode(Wv, mode, n) print(f" {mode:22}{r['全滅率']*100:8.1f}%{r['到達']*100:12.1f}%" f"{r['手取り']:+14,.0f}{r['プラス率']*100:9.1f}%{r['SD']:12,.0f}") print("\n" + "=" * 104) print(" 【B】『過去ベストで選ぶ』と、選んだ波どうしの相関が上がる") print("=" * 104) for k in (5, 10, 20): a, b = selection_rho(Wv, k) print(f" {k:2d}波 … 無作為 ρ={a:+.3f} / 過去ベスト ρ={b:+.3f} （差 {b-a:+.3f}）") print("\n" + "=" * 104) print(" 【C】相関 ρ のとき、リスクはどこまで下がるか √((1+(N-1)ρ)/N)") print("=" * 104) print(f" {'ρ':>7}{'N=5':>9}{'N=10':>9}{'N=20':>9}{'N=100':>9}{'下限 √ρ':>11}") for rho in (0.0, 0.007, 0.1, 0.3, 0.5, 0.8): row = "".join(f"{risk_ratio(rho, n):>9.3f}" for n in (5, 10, 20, 100)) print(f" {rho:>7.3f}{row}{np.sqrt(rho):>11.3f}")