# -*- coding: utf-8 -*- """「全口座バラバラのプロップ」と「自己資金トレーダー」を、同じ現金・同じ戦略で比べる。 比べ方: 同じ現金 C を A … 参加費に投下する（$500 × N口座、全部バラバラの波、口座数には上限あり） B … そのまま自己資金として運用する（同じ波を等ウェイトで持つ） 戦略も期間も相場もまったく同じにして、終端資産を比べる。 肝は2つ: ・A は「損失が参加費で打ち止めのレバレッジ」を買っている ・A は投下できる金額に上限がある（口座数の上限）。B に上限はない → だから資金量のどこかで逆転する。その交差点を実測する。 使い方: python scripts/fetch_wave_data.py python scripts/sim-prop-vs-own-capital.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 = 20260918, 3000 FEE, ACCOUNT, SPLIT = 500.0, 100_000.0, 0.80 P1, P2, MAX_DD, DAILY_DD = 0.08, 0.05, 0.10, 0.05 MAX_ACCOUNTS = 40 # 1人が現実的に持てるチャレンジ口座数の上限 YEAR = 260 OWN_RUIN = 0.20 # 自己資金は元本の2割まで減ったら実質再起不能とみなす OWN_WAVES = 10 # 自己資金側も同じだけ分散させる（器の違いだけを見るため） def run_prop(Wv, starts, cash, years, rng, jump_q=0.0, jump_size=0.0): """A: 参加費に投下。年ごとに、手元の現金で買えるだけ買い直す（焚き火）。""" T, M = len(starts), Wv.shape[1] wallet = np.full(T, float(cash)) for y in range(years): n = np.minimum((wallet // FEE).astype(int), MAX_ACCOUNTS) n_max = int(n.max()) if n_max == 0: break waves = np.array([rng.choice(M, size=n_max, replace=False) for _ in range(T)]) off = starts + y * YEAR R = np.empty((T, YEAR, n_max)) for i, s in enumerate(off): R[i] = Wv[s:s + YEAR][:, waves[i]] R = apply_jumps(R, off, waves, jump_q, jump_size, rng, YEAR) live = np.arange(n_max)[None, :] < n[:, None] # 買えた口座だけ有効 wallet -= FEE * n eq = np.ones((T, n_max)); alive = live.copy(); stage = np.zeros((T, n_max), int) tgt = np.array([P1, P2, np.inf]) for d in range(YEAR): step = R[:, d, :] eq = eq + step * alive 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) wallet += (np.where(funded, np.maximum(eq - 1.0, 0.0), 0.0) * ACCOUNT * SPLIT).sum(1) return wallet # --- 一撃（介入・フラッシュクラッシュ）のモデル --- # 実際の一撃は「ある1銘柄」が動くもので、損得は各戦略の建玉の向きで分かれる。 # 全戦略を同時に逆側で殴るモデルにすると、分散の効果を消してしまい結論が壊れる。 _EXPO = {"E": None, "sym_id": None, "n_sym": 0} def _ensure_expo(): if _EXPO["E"] is None: w, e, syms = build_waves(with_exposure=True) u = sorted(set(syms)) _EXPO["E"] = e.values _EXPO["sym_id"] = np.array([u.index(x) for x in syms]) _EXPO["n_sym"] = len(u) return _EXPO def apply_jumps(R, starts, waves, q, size, rng, days): """R に一撃を加える。q=発生確率/日、size=その銘柄が動く割合。""" if q <= 0 or size <= 0: return R ex = _ensure_expo() E, sid, ns = ex["E"], ex["sym_id"], ex["n_sym"] hit = rng.random((len(starts), days)) < q tgt = rng.integers(0, ns, size=(len(starts), days)) sgn = rng.choice([-1.0, 1.0], size=(len(starts), days)) for i in range(len(starts)): for d in np.nonzero(hit[i])[0]: m = sid[waves[i]] == tgt[i, d] if m.any(): R[i, d, m] += E[starts[i] + d][waves[i]][m] * sgn[i, d] * size return R def run_own(Wv, starts, cash, years, lev, rng, alpha=0.0, regime_len=0, jump_q=0.0, jump_size=0.0): """B: 自己資金。同じ波を OWN_WAVES 本の等ウェイトで持ち、レバ lev をかける。""" T, M = len(starts), Wv.shape[1] sel = np.array([rng.choice(M, size=OWN_WAVES, replace=False) for _ in range(T)]) days = years * YEAR R = np.empty((T, days, OWN_WAVES)) for i, s in enumerate(starts): R[i] = Wv[s:s + days][:, sel[i]] R = apply_jumps(R, starts, sel, jump_q, jump_size, rng, days) port = R.mean(2) # 等ウェイトの合成 # 間欠エッジ: regime_len 日ごとに α が ON/OFF する。期間平均の α は同じ。 if regime_len > 0: blocks = (np.arange(days) // regime_len) % 2 a_seq = np.where(blocks == 0, 2.0 * alpha, 0.0) else: a_seq = np.full(days, alpha) eq = np.ones(T); alive = np.ones(T, bool) for d in range(days): step = port[:, d] + a_seq[d] eq = np.where(alive, eq * (1.0 + lev * step), eq) # 自己資金は複利 alive &= eq > OWN_RUIN return np.where(alive, eq, OWN_RUIN) * cash, ~alive def summarize(final, cash): return dict(期待=final.mean(), 中央=np.median(final), プラス率=(final > cash).mean(), 倍率=final.mean() / cash) if __name__ == "__main__": W = build_waves(); Wv = W.values print(f"波 {Wv.shape[1]}本 / {W.index[0].date()}〜{W.index[-1].date()}") print(f"A=参加費${FEE:,.0f}・口座上限{MAX_ACCOUNTS} / B=自己資金・{OWN_WAVES}波の等ウェイト・複利") print(f"トライアル {TRIALS:,}・実カレンダー上の開始日をランダム抽出") for years in (1, 3): days = years * YEAR rng0 = np.random.default_rng(SEED) starts = rng0.integers(0, len(Wv) - days - 1, size=TRIALS) print("\n" + "=" * 100) print(f" 【{years}年】同じ現金を、参加費に入れるか・自分で運用するか") print("=" * 100) print(f" {'元手':>10}{'口座数':>7} {'A プロップ':>28} {'B 自己資金（レバ1倍）':>30}") print(f" {'':>10}{'':>7} {'期待終端':>12}{'中央値':>10}{'プラス率':>8}" f" {'期待終端':>12}{'中央値':>10}{'プラス率':>8}") for cash in (2_500, 10_000, 50_000, 200_000, 1_000_000): a = run_prop(Wv, starts, cash, years, np.random.default_rng(SEED)) b, ruin = run_own(Wv, starts, cash, years, 1.0, np.random.default_rng(SEED)) sa, sb = summarize(a, cash), summarize(b, cash) n = min(int(cash // FEE), MAX_ACCOUNTS) print(f" ${cash:>9,.0f}{n:>7} ${sa['期待']:>11,.0f}${sa['中央']:>9,.0f}" f"{sa['プラス率']*100:>7.1f}% ${sb['期待']:>11,.0f}${sb['中央']:>9,.0f}{sb['プラス率']*100:>7.1f}%") # --- 感応度: 自己資金がレバをどこまで上げたらプロップに並ぶか --- print("\n" + "=" * 100) print(" 【感応度】自己資金がレバレッジを上げていくと（元手$10,000・1年）") print("=" * 100) rng0 = np.random.default_rng(SEED) starts = rng0.integers(0, len(Wv) - YEAR - 1, size=TRIALS) cash = 10_000 a = run_prop(Wv, starts, cash, 1, np.random.default_rng(SEED)) sa = summarize(a, cash) print(f" A プロップ（{min(int(cash//FEE),MAX_ACCOUNTS)}口座）… 期待終端 ${sa['期待']:,.0f}" f"（{sa['倍率']:.2f}倍）/ 中央値 ${sa['中央']:,.0f} / プラス率 {sa['プラス率']*100:.1f}%") print(f"\n {'レバ':>6}{'期待終端':>14}{'倍率':>8}{'中央値':>12}{'プラス率':>10}{'破産率':>9}") for lev in (1, 2, 5, 10, 20, 40, 60): b, ruin = run_own(Wv, starts, cash, 1, float(lev), np.random.default_rng(SEED)) sb = summarize(b, cash) mark = " ← Aを超える" if sb["期待"] > sa["期待"] else "" print(f" {lev:>5}倍${sb['期待']:>13,.0f}{sb['倍率']:>8.2f}${sb['中央']:>11,.0f}" f"{sb['プラス率']*100:>9.1f}%{ruin.mean()*100:>8.1f}%{mark}") # --- 感応度2: 自己資金の「腕」をどこまで上げたらプロップに勝てるか --- # 平均ではなく中央値で比べる。1回しか試行できない人が実際に体験するのは中央値のほうで、 # 高レバの複利は平均だけを発散させて実態を見えなくするため。 # レバレッジは現実的な上限（20倍）までとする。 print("\n" + "=" * 100) print(" 【感応度2】自己資金の腕をどこまで上げればプロップに並ぶか（元手$10,000・1年）") print(" ※ 各腕について、中央値が最大になるレバレッジを選んでいる（レバは20倍まで）") print("=" * 100) base = np.array([Wv[s0:s0 + YEAR].mean(1) for s0 in starts]) print(f" ベースの10波ポートフォリオの年率Sharpe ≒ {base.mean()/base.std()*np.sqrt(252):.2f}") print(f" A プロップ … 期待終端 ${sa['期待']:,.0f} / 中央値 ${sa['中央']:,.0f}\n") print(f" {'上乗せα(日次)':>14}{'実効Sharpe':>12}{'最良レバ':>10}" f"{'中央値':>12}{'期待終端':>14}{'プラス率':>10}{'破産率':>9} 中央値で判定") for alpha in (0.0, 0.00003, 0.00005, 0.00008, 0.00010, 0.00013, 0.00020): best = None for lev in (1, 2, 3, 4, 5, 6, 8, 10, 12, 16, 20): b, ruin = run_own(Wv, starts, cash, 1, float(lev), np.random.default_rng(SEED), alpha=alpha) sb = summarize(b, cash) if best is None or sb["中央"] > best[1]["中央"]: best = (lev, sb, ruin.mean()) lev, sb, rn = best eff = (base.mean() + alpha) / base.std() * np.sqrt(252) judge = "★Aを超える" if sb["中央"] > sa["中央"] else "届かない" print(f" {alpha*100:>13.2f}%{eff:>12.2f}{lev:>9}倍${sb['中央']:>11,.0f}" f"${sb['期待']:>13,.0f}{sb['プラス率']*100:>9.1f}%{rn*100:>8.1f}% {judge}") # --- 感応度3: 一撃（介入・フラッシュクラッシュ）に効くのは何か --- print("\n" + "=" * 100) print(" 【感応度3】レバ20倍の自己資金トレーダーに、年1回・4%の一撃が来たら") print(" ※ 一撃は『ある1銘柄が動く』もので、損得は各戦略の建玉の向きで分かれる") print("=" * 100) print(f" {'戦略の本数':>12}{'一撃なし中央値':>17}{'一撃あり中央値':>17}{'破産率':>9}") saved = OWN_WAVES for nw in (1, 2, 3, 5, 10): globals()["OWN_WAVES"] = nw b0, _ = run_own(Wv, starts, cash, 1, 20.0, np.random.default_rng(SEED), alpha=0.00013) b1, r1 = run_own(Wv, starts, cash, 1, 20.0, np.random.default_rng(SEED), alpha=0.00013, jump_q=1/252, jump_size=0.04) print(f" {nw:>10}本${np.median(b0):>16,.0f}${np.median(b1):>16,.0f}{r1.mean()*100:>8.1f}%") globals()["OWN_WAVES"] = saved print("\n → 破産を決めているのは一撃の有無ではなく、分散の本数とレバレッジ")