※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。
本記事は実装方法の解説であり、特定の戦略やパラメータの収益性を主張するものではありません。掲載するコードはすべて記事31のバックテスト基盤と記事32の検証手法を通す前提で書かれています。過去の結果は将来を保証しません。
| 理由 | 内容 |
|---|---|
| 学術的裏付け | Moskowitz, Ooi & Pedersen (2012) は58の先物・通貨で1〜12か月の時系列モメンタムを確認。Hurst et al. (2017) は100年以上のデータで再現 |
| パラメータが少ない | ルックバック期間とトレール幅の2つで済む。過剰最適化しにくい（記事32のDSRを通しやすい） |
| 損益分布が有利 | 勝率は低い（30〜45%）が、利益が損失より大きい。少数の大勝ちが全体を支える |
| 実装が単純 | 確定足だけで判断でき、低遅延を要求しない。マーケットメイク（記事37）と違いレイテンシ競争にならない |
トレンドフォローは「負けが続き、稀に大きく勝つ」戦略です。勝率30%台は正常で、連続10回の負けも普通に起こります。そして年間利益の大半が数回のトレードから生まれます。
問題は心理面です。連敗中に「壊れた」と判断して止めると、その直後の大勝ちを取り逃がし、損失だけが確定します。実際にこれで失敗する人が多い。
だから連敗の分布を事前に把握しておくことが実装の一部です（サンプル7で最長連敗を計測します）。「何連敗までは正常か」を数値で知っていれば、耐えられます。
"""signals.py — トレンドフォローの3種類のシグナル
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
いずれも「確定足の情報だけ」で計算する。
未来を参照しないことは記事31のルックアヘッドテストで検証する。
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import IntEnum
import numpy as np
import pandas as pd
class Direction(IntEnum):
SHORT = -1
FLAT = 0
LONG = 1
@dataclass(frozen=True)
class SignalSpec:
name: str
lookback: int
description: str
# ---- 1) 移動平均クロス -------------------------------------------------
def ma_cross_signal(close: pd.Series, fast: int, slow: int) -> pd.Series:
"""速いMAが遅いMAを上回っていればロング。最も単純な形。
注意: shift(1)は呼び出し側で行う（この関数は「その足時点の状態」を返す）。
"""
if fast >= slow:
raise ValueError("fast < slow が必要です")
f = close.rolling(fast, min_periods=fast).mean()
s = close.rolling(slow, min_periods=slow).mean()
sig = pd.Series(Direction.FLAT, index=close.index, dtype="int8")
sig[f > s] = Direction.LONG
sig[f < s] = Direction.SHORT
return sig
# ---- 2) ドンチャンブレイクアウト ----------------------------------------
def donchian_signal(high: pd.Series, low: pd.Series, close: pd.Series,
entry_period: int = 55,
exit_period: int = 20) -> pd.Series:
"""N期間高値を超えたらロング、N期間安値を割ったらショート。
タートルズ流。エントリーとエグジットで期間を分けるのが要点で、
エグジットを短くすると利益の保持と損失の切り離しが両立する。
重要: 高値・安値は「1本前まで」で計算する（当日の高値を含めると
未来参照になる）。shift(1)を内部で入れている。
"""
upper = high.rolling(entry_period, min_periods=entry_period).max().shift(1)
lower = low.rolling(entry_period, min_periods=entry_period).min().shift(1)
exit_up = high.rolling(exit_period, min_periods=exit_period).max().shift(1)
exit_dn = low.rolling(exit_period, min_periods=exit_period).min().shift(1)
sig = pd.Series(Direction.FLAT, index=close.index, dtype="int8")
state = Direction.FLAT
values = np.zeros(len(close), dtype=np.int8)
c = close.to_numpy()
up, dn = upper.to_numpy(), lower.to_numpy()
xu, xd = exit_up.to_numpy(), exit_dn.to_numpy()
for i in range(len(c)):
if np.isnan(up[i]) or np.isnan(dn[i]):
values[i] = Direction.FLAT
continue
if state == Direction.FLAT:
if c[i] > up[i]:
state = Direction.LONG
elif c[i] < dn[i]:
state = Direction.SHORT
elif state == Direction.LONG:
# エグジット期間の安値を割ったら撤退
if not np.isnan(xd[i]) and c[i] < xd[i]:
state = Direction.FLAT
if c[i] < dn[i]:
state = Direction.SHORT # ドテン
else: # SHORT
if not np.isnan(xu[i]) and c[i] > xu[i]:
state = Direction.FLAT
if c[i] > up[i]:
state = Direction.LONG
values[i] = state
sig[:] = values
return sig
# ---- 3) 時系列モメンタム（TSMOM） --------------------------------------
def tsmom_signal(close: pd.Series, lookback: int = 168,
threshold: float = 0.0) -> pd.Series:
"""過去N期間のリターンの符号でポジションを決める。
Moskowitz, Ooi & Pedersen (2012) の最も単純な形。
「上がってきたものは上がり続ける」という時系列方向の慣性を使う。
threshold を上げると、弱いトレンドを無視できる（ノイズ低減）。
"""
ret = close.pct_change(lookback)
sig = pd.Series(Direction.FLAT, index=close.index, dtype="int8")
sig[ret > threshold] = Direction.LONG
sig[ret < -threshold] = Direction.SHORT
return sig
def tsmom_multi_horizon(close: pd.Series,
lookbacks: tuple[int, ...] = (24, 72, 168, 336),
min_agreement: float = 0.5) -> pd.Series:
"""複数のルックバックの合意を取る。単一期間への依存を減らす。
min_agreement=0.5 なら、過半数が同じ方向を示したときだけポジションを取る。
パラメータ1つに賭けないため、過剰最適化しにくい。
"""
votes = pd.DataFrame({
f"h{lb}": tsmom_signal(close, lb).astype(float)
for lb in lookbacks
})
score = votes.mean(axis=1) # -1 〜 +1
sig = pd.Series(Direction.FLAT, index=close.index, dtype="int8")
sig[score >= min_agreement] = Direction.LONG
sig[score <= -min_agreement] = Direction.SHORT
return sig
# ---- 比較 ---------------------------------------------------------------
def compare_signals(bars: pd.DataFrame) -> pd.DataFrame:
"""各シグナルの性質を比較する（回転率・ポジション保有率）。"""
close, high, low = bars["close"], bars["high"], bars["low"]
sigs = {
"MA(20,60)": ma_cross_signal(close, 20, 60),
"Donchian(55,20)": donchian_signal(high, low, close, 55, 20),
"TSMOM(168)": tsmom_signal(close, 168),
"TSMOM(multi)": tsmom_multi_horizon(close),
}
rows = []
for name, sig in sigs.items():
s = sig.dropna()
if s.empty:
continue
changes = (s.diff() != 0).sum()
rows.append({
"signal": name,
"ロング率": float((s == Direction.LONG).mean()),
"ショート率": float((s == Direction.SHORT).mean()),
"フラット率": float((s == Direction.FLAT).mean()),
"方向転換回数": int(changes),
"平均保有本数": float(len(s) / max(changes, 1)),
})
return pd.DataFrame(rows).set_index("signal")
if __name__ == "__main__":
from kline_store import KlineStore
bars = KlineStore().load("BTCUSDT", "60")
if bars.empty:
raise SystemExit("先に記事07でデータを取得してください")
bars = bars.loc["2024-01-01":]
print(f"{len(bars)}本の1時間足で比較\n")
with pd.option_context("display.float_format", lambda v: f"{v:.3f}"):
print(compare_signals(bars).to_string())
print("\n--- 読み方 ---")
print(" 方向転換回数が多い = 回転率が高い = 手数料負担が大きい")
print(" 平均保有本数が短いシグナルは、往復コストを回収できるか要確認")
print(" （1時間足で平均20本保有なら、20時間で往復コストを回収する必要）")単純なトレンドフォローの最大の弱点は、銘柄・時期によってリスク量がばらつくことです。BTCとDOGEを同じ枚数持てばリスクは全く違い、同じBTCでもボラが2倍になればリスクも2倍です。
ボラティリティ正規化は、これを「目標リスク量」で揃えます。TSMOMの論文でも中核的な要素です。
"""vol_target.py — ボラティリティ正規化によるポジションサイジング
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
考え方:
目標年率ボラ / 銘柄の年率ボラ = 想定元本の倍率
ボラが高い銘柄・時期には小さく、低いときは大きく持つ。
これにより「常に同じリスク量」を維持できる。
参考: Moskowitz, Ooi & Pedersen (2012) の TSMOM 構成、
Moreira & Muir (2017) "Volatility-Managed Portfolios"
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
import numpy as np
import pandas as pd
BARS_PER_YEAR = {"1": 525_600, "5": 105_120, "15": 35_040, "60": 8_760,
"240": 2_190, "D": 365}
@dataclass(frozen=True)
class VolTargetConfig:
target_annual_vol: float = 0.20 # 目標年率ボラ 20%
vol_lookback: int = 168 # ボラ推定期間（1時間足で1週間）
max_leverage: float = 3.0 # 実効レバレッジの上限
min_leverage: float = 0.0
vol_floor: float = 0.05 # 年率ボラの下限（ゼロ割り防止）
def realised_vol(close: pd.Series, lookback: int,
bars_per_year: int) -> pd.Series:
"""対数リターンの標準偏差から年率ボラを推定する。"""
log_ret = np.log(close / close.shift(1))
return log_ret.rolling(lookback, min_periods=max(lookback // 2, 10)).std() \
* np.sqrt(bars_per_year)
def yang_zhang_vol(bars: pd.DataFrame, lookback: int,
bars_per_year: int) -> pd.Series:
"""Yang-Zhang推定量。始値ギャップとレンジを使い、終値ベースより効率的。
参考: Yang & Zhang (2000)
"""
o, h, l, c = (bars["open"], bars["high"], bars["low"], bars["close"])
c_prev = c.shift(1)
# 翌日への持ち越し（オーバーナイト）成分
log_oc_prev = np.log(o / c_prev)
sigma_o = log_oc_prev.rolling(lookback).var()
# 日中の終値成分
log_co = np.log(c / o)
sigma_c = log_co.rolling(lookback).var()
# Rogers-Satchell成分（レンジを使う）
rs = (np.log(h / c) * np.log(h / o) + np.log(l / c) * np.log(l / o))
sigma_rs = rs.rolling(lookback).mean()
k = 0.34 / (1.34 + (lookback + 1) / (lookback - 1))
var = sigma_o + k * sigma_c + (1 - k) * sigma_rs
return np.sqrt(var.clip(lower=0)) * np.sqrt(bars_per_year)
def vol_scaled_leverage(vol: pd.Series,
cfg: VolTargetConfig | None = None) -> pd.Series:
"""年率ボラから、想定元本の倍率（実効レバレッジ）を求める。"""
cfg = cfg or VolTargetConfig()
safe_vol = vol.clip(lower=cfg.vol_floor)
lev = cfg.target_annual_vol / safe_vol
return lev.clip(lower=cfg.min_leverage, upper=cfg.max_leverage)
def position_notional(equity: Decimal, leverage: float,
direction: int) -> Decimal:
"""純資産と倍率から、目標の想定元本（符号付き）を返す。"""
if direction == 0 or leverage <= 0:
return Decimal(0)
return equity * Decimal(str(round(leverage, 6))) * direction
def compare_sizing(bars: pd.DataFrame, interval: str = "60") -> pd.DataFrame:
"""固定サイズとボラ正規化を比較する。"""
bpy = BARS_PER_YEAR.get(interval, 8_760)
cfg = VolTargetConfig()
close_vol = realised_vol(bars["close"], cfg.vol_lookback, bpy)
yz_vol = yang_zhang_vol(bars, cfg.vol_lookback, bpy)
out = pd.DataFrame({
"close_vol": close_vol,
"yz_vol": yz_vol,
"lev_close": vol_scaled_leverage(close_vol, cfg),
"lev_yz": vol_scaled_leverage(yz_vol, cfg),
}).dropna()
# 実現するリスク量（倍率 × 実ボラ）が目標に近いかを確認
out["realised_risk_close"] = out["lev_close"] * out["close_vol"]
out["realised_risk_yz"] = out["lev_yz"] * out["yz_vol"]
return out
if __name__ == "__main__":
from kline_store import KlineStore
store = KlineStore()
cfg = VolTargetConfig()
print(f"目標年率ボラ {cfg.target_annual_vol:.0%} / "
f"上限レバレッジ {cfg.max_leverage}x\n")
print(f"{'symbol':<11}{'年率ボラ':>10}{'倍率':>8}"
f"{'実現リスク':>11}{'ボラの変動幅':>14}")
print("-" * 56)
for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT"):
bars = store.load(sym, "60")
if bars.empty:
continue
bars = bars.loc["2024-01-01":]
df = compare_sizing(bars)
if df.empty:
continue
vol = df["yz_vol"].iloc[-1]
lev = df["lev_yz"].iloc[-1]
risk = df["realised_risk_yz"].mean()
vol_range = f"{df['yz_vol'].min():.0%}〜{df['yz_vol'].max():.0%}"
print(f"{sym:<11}{vol:>9.1%}{lev:>8.2f}x{risk:>10.1%}"
f"{vol_range:>14}")
print("\n※ 銘柄ごとにボラが数倍違うため、同じ枚数を持つとリスクが揃わない。")
print(" ボラ正規化すれば『実現リスク』が目標付近に揃う。")
print(" 上限レバレッジでクリップされる分、完全には一致しない。")シグナルを変えるより、ボラ正規化を入れるほうが成績への影響が大きいことが多いです。理由は、(1) 銘柄間のリスクが揃うため分散が効く、(2) 高ボラ局面で自動的にポジションを縮めるため大きなドローダウンを避けられる、(3) パラメータを増やさない（目標ボラ1つ）。
特に(3)が重要です。記事32で見たように、パラメータを増やすほどDSRが厳しくなります。ボラ正規化は「経済的に正しい」変更なので、データに合わせて選んだパラメータではありません。
「+2%で利確」のような固定利確は、トレンドフォローの収益構造を壊します。この戦略は「少数の大勝ちが全体を支える」ため、その大勝ちを早期に切ると期待値がマイナスになります。
正しいのは「損切りは固定（または小さめ）、利確はトレールで伸ばす」非対称な設計です。ATRトレールなら、ボラティリティに応じて追従幅が自動調整されます。
"""trail.py — ATRベースのトレーリングストップ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
設計:
- 損切り: エントリー時のATR × k（固定）
- トレール: 有利方向にのみ動かす（不利方向には絶対に動かさない）
- 利確: 設定しない（トレールに任せる）
"""
from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np
import pandas as pd
def atr(bars: pd.DataFrame, period: int = 14) -> pd.Series:
"""ATR（Average True Range）。TA-Lib不要の実装。"""
h, l, c = bars["high"], bars["low"], bars["close"]
prev_c = c.shift(1)
tr = pd.concat([h - l, (h - prev_c).abs(), (l - prev_c).abs()],
axis=1).max(axis=1)
return tr.rolling(period, min_periods=period).mean()
def chandelier_exit(bars: pd.DataFrame, period: int = 22,
mult: float = 3.0) -> pd.DataFrame:
"""シャンデリア・エグジット。高値からATR×kを引いた水準。
ロングの損切り = 直近N期間の高値 − ATR × mult
ショートの損切り = 直近N期間の安値 + ATR × mult
注意: shift(1)で「1本前まで」の情報にする（未来参照の防止）。
"""
a = atr(bars, period)
highest = bars["high"].rolling(period, min_periods=period).max()
lowest = bars["low"].rolling(period, min_periods=period).min()
return pd.DataFrame({
"long_stop": (highest - a * mult).shift(1),
"short_stop": (lowest + a * mult).shift(1),
"atr": a.shift(1),
}, index=bars.index)
@dataclass
class TrailState:
"""1ポジションのトレール状態。単調性（有利方向のみ）を保証する。"""
is_long: bool
entry_price: float
stop_price: float
atr_at_entry: float
best_price: float
updates: int = 0
history: list[float] = field(default_factory=list)
def update(self, price: float, atr_now: float, mult: float) -> bool:
"""価格とATRから損切り水準を更新する。改善したらTrueを返す。"""
if self.is_long:
self.best_price = max(self.best_price, price)
candidate = self.best_price - atr_now * mult
improved = candidate > self.stop_price
else:
self.best_price = min(self.best_price, price)
candidate = self.best_price + atr_now * mult
improved = candidate < self.stop_price
if improved:
self.stop_price = candidate
self.updates += 1
self.history.append(candidate)
return improved
def is_hit(self, price: float) -> bool:
return (price <= self.stop_price if self.is_long
else price >= self.stop_price)
@property
def risk_pct(self) -> float:
"""現在の損切りまでの距離（エントリー価格に対する%）。"""
return abs(self.stop_price - self.entry_price) / self.entry_price * 100
@property
def locked_pct(self) -> float:
"""確定させた利益（損切りがエントリーを超えた分）。"""
if self.is_long:
gain = self.stop_price - self.entry_price
else:
gain = self.entry_price - self.stop_price
return gain / self.entry_price * 100
def make_trail(is_long: bool, entry_price: float, atr_value: float,
mult: float = 3.0) -> TrailState:
stop = (entry_price - atr_value * mult if is_long
else entry_price + atr_value * mult)
return TrailState(is_long=is_long, entry_price=entry_price,
stop_price=stop, atr_at_entry=atr_value,
best_price=entry_price)
def simulate_trail(bars: pd.DataFrame, is_long: bool = True,
period: int = 22, mult: float = 3.0) -> pd.DataFrame:
"""トレールの挙動を1トレード分シミュレートする（挙動確認用）。"""
a = atr(bars, period).shift(1)
valid = bars.loc[a.notna()]
if valid.empty:
return pd.DataFrame()
entry_idx = valid.index[0]
entry = float(valid["close"].iloc[0])
st = make_trail(is_long, entry, float(a.loc[entry_idx]), mult)
rows = []
for ts, row in valid.iterrows():
atr_now = float(a.loc[ts])
# 不利方向の判定を先に行う（同じ足で両方触れた場合の保守的な扱い）
worst = float(row["low"] if is_long else row["high"])
if st.is_hit(worst):
rows.append({"ts": ts, "price": worst, "stop": st.stop_price,
"hit": True, "locked_pct": st.locked_pct})
break
st.update(float(row["close"]), atr_now, mult)
rows.append({"ts": ts, "price": float(row["close"]),
"stop": st.stop_price, "hit": False,
"locked_pct": st.locked_pct})
return pd.DataFrame(rows).set_index("ts")
if __name__ == "__main__":
from kline_store import KlineStore
bars = KlineStore().load("BTCUSDT", "60")
if bars.empty:
raise SystemExit("先に記事07でデータを取得してください")
bars = bars.loc["2025-01-01":"2025-03-01"]
print("=== ATRトレールの挙動（mult別） ===")
print(f"{'mult':>6}{'保有本数':>10}{'確定利益%':>12}{'初期リスク%':>13}")
print("-" * 42)
for mult in (1.5, 2.0, 3.0, 4.0, 6.0):
sim = simulate_trail(bars, is_long=True, mult=mult)
if sim.empty:
continue
print(f"{mult:>6.1f}{len(sim):>10}"
f"{sim['locked_pct'].iloc[-1]:>+11.2f}%"
f"{abs(sim['locked_pct'].iloc[0]):>12.2f}%")
print("\n※ multが小さいと早く切られ、大きいと伸ばせるが損失も大きい。")
print(" トレンドフォローでは3.0前後から始めるのが一般的だが、")
print(" 記事32のWFAで検証してから決めること。")
ce = chandelier_exit(bars, 22, 3.0)
print(f"\n直近のシャンデリア水準:")
print(f" ロングの損切り : {ce['long_stop'].iloc[-1]:,.1f}")
print(f" ショートの損切り: {ce['short_stop'].iloc[-1]:,.1f}")
print(f" ATR(22) : {ce['atr'].iloc[-1]:,.1f}")トレンドフォローはレンジ相場で損失を積み上げます（ダマシのブレイクを繰り返す）。「トレンドが出ているか」を判定して、出ていないときは休むのが有効です。
"""regime.py — トレンド相場かレンジ相場かを判定するフィルタ
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
注意: フィルタを増やすほどパラメータが増え、過剰最適化しやすくなる。
「効いた気がする」で追加せず、記事32のWFA/DSRで検証すること。
"""
from __future__ import annotations
import numpy as np
import pandas as pd
def efficiency_ratio(close: pd.Series, period: int = 100) -> pd.Series:
"""Kaufmanの効率比。方向性のある動きの割合を測る。
(期間の純変化) / (期間内の変動の合計)
1に近い = 一直線に動いている（トレンド）
0に近い = 行ったり来たり（レンジ）
"""
net = (close - close.shift(period)).abs()
path = close.diff().abs().rolling(period).sum()
return (net / path).replace([np.inf, -np.inf], np.nan)
def adx(bars: pd.DataFrame, period: int = 14) -> pd.Series:
"""ADX（Average Directional Index）。トレンドの強さを測る定番指標。
25以上でトレンドあり、20以下でレンジとされるのが一般的。
"""
h, l, c = bars["high"], bars["low"], bars["close"]
up = h.diff()
dn = -l.diff()
plus_dm = np.where((up > dn) & (up > 0), up, 0.0)
minus_dm = np.where((dn > up) & (dn > 0), dn, 0.0)
prev_c = c.shift(1)
tr = pd.concat([h - l, (h - prev_c).abs(), (l - prev_c).abs()],
axis=1).max(axis=1)
atr_ = tr.rolling(period).mean()
plus_di = 100 * pd.Series(plus_dm, index=bars.index).rolling(period).mean() / atr_
minus_di = 100 * pd.Series(minus_dm, index=bars.index).rolling(period).mean() / atr_
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
return dx.rolling(period).mean()
def higher_timeframe_filter(bars: pd.DataFrame, htf: str = "4h",
ma_period: int = 50) -> pd.Series:
"""上位時間足のトレンド方向。順張り方向を制限する。
1時間足で売買しつつ、4時間足が上向きならロングのみ、
という使い方をする。
"""
htf_close = bars["close"].resample(htf, label="left",
closed="left").last().dropna()
htf_ma = htf_close.rolling(ma_period, min_periods=ma_period).mean()
direction = pd.Series(0, index=htf_close.index, dtype="int8")
direction[htf_close > htf_ma] = 1
direction[htf_close < htf_ma] = -1
# 上位足の値を下位足に前方埋めする（未来を見ないようshift）
return direction.shift(1).reindex(bars.index, method="ffill").fillna(0)
def vol_regime(bars: pd.DataFrame, lookback: int = 168,
percentile_window: int = 720) -> pd.Series:
"""ボラティリティの相対水準（0〜1）。極端に高い局面を避けるのに使う。"""
from vol_target import realised_vol
vol = realised_vol(bars["close"], lookback, 8_760)
return vol.rolling(percentile_window, min_periods=lookback).rank(pct=True)
def build_filter(bars: pd.DataFrame, *,
min_efficiency: float | None = 0.25,
min_adx: float | None = 20.0,
use_htf: bool = True,
max_vol_percentile: float | None = 0.95) -> pd.DataFrame:
"""複数のフィルタをまとめて評価する。
戻り値の allow_long / allow_short が True の足でのみエントリーする。
"""
out = pd.DataFrame(index=bars.index)
out["allow_long"] = True
out["allow_short"] = True
if min_efficiency is not None:
er = efficiency_ratio(bars["close"]).shift(1)
ok = er >= min_efficiency
out["allow_long"] &= ok.fillna(False)
out["allow_short"] &= ok.fillna(False)
out["efficiency"] = er
if min_adx is not None:
a = adx(bars).shift(1)
ok = a >= min_adx
out["allow_long"] &= ok.fillna(False)
out["allow_short"] &= ok.fillna(False)
out["adx"] = a
if use_htf:
htf = higher_timeframe_filter(bars)
out["allow_long"] &= (htf >= 0)
out["allow_short"] &= (htf <= 0)
out["htf"] = htf
if max_vol_percentile is not None:
vp = vol_regime(bars).shift(1)
ok = vp <= max_vol_percentile
out["allow_long"] &= ok.fillna(True)
out["allow_short"] &= ok.fillna(True)
out["vol_pct"] = vp
return out
if __name__ == "__main__":
from kline_store import KlineStore
bars = KlineStore().load("BTCUSDT", "60")
if bars.empty:
raise SystemExit("先に記事07でデータを取得してください")
bars = bars.loc["2024-01-01":]
f = build_filter(bars)
print(f"全 {len(f)}本のうち")
print(f" ロング可 : {f['allow_long'].mean():.1%}")
print(f" ショート可 : {f['allow_short'].mean():.1%}")
print(f" どちらも不可 : "
f"{(~f['allow_long'] & ~f['allow_short']).mean():.1%}")
print("\n--- 各フィルタ単体の通過率 ---")
for name, kw in (
("効率比のみ", dict(min_adx=None, use_htf=False,
max_vol_percentile=None)),
("ADXのみ", dict(min_efficiency=None, use_htf=False,
max_vol_percentile=None)),
("上位足のみ", dict(min_efficiency=None, min_adx=None,
max_vol_percentile=None)),
("ボラ上限のみ", dict(min_efficiency=None, min_adx=None,
use_htf=False)),
):
sub = build_filter(bars, **kw)
print(f" {name:<14} ロング可 {sub['allow_long'].mean():>6.1%}")
print("\n※ フィルタは取引機会を減らす。減った分だけ約定回数が落ち、")
print(" 統計的な信頼性も下がる（記事32の min_trades を下回りやすい）。")
print(" 『フィルタで成績が良くなった』は過剰最適化の典型なので、")
print(" 必ずWFAとDSRで検証すること。")ここまでの部品を、記事31の Strategy 基底クラスに統合します。これ1つでバックテストと本番の両方が動きます（記事30）。
"""strategies/trend_follow.py — トレンドフォロー戦略（完成版）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
このクラスは記事31のバックテストエンジンと記事30の本番Runnerの
両方でそのまま動く。環境を意識するコードは一切含まない。
注意: 収益性を主張するものではなく、実装例です。
使用前に記事32の検証を必ず通してください。
"""
from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np
import pandas as pd
from bt.strategy import Strategy
from bt.types import Fill, Order, OrderType, Position, Side
from regime import build_filter
from signals import Direction, donchian_signal, tsmom_multi_horizon
from trail import TrailState, atr, make_trail
from vol_target import VolTargetConfig, realised_vol, vol_scaled_leverage
@dataclass
class TrendConfig:
# --- シグナル（パラメータは少なく保つ） ---
signal: str = "donchian" # donchian / tsmom
entry_period: int = 55
exit_period: int = 20
tsmom_lookbacks: tuple[int, ...] = (24, 72, 168, 336)
# --- サイジング ---
target_annual_vol: float = 0.20
vol_lookback: int = 168
max_leverage: float = 2.0
# --- トレール ---
atr_period: int = 22
atr_mult: float = 3.0
# --- フィルタ（Noneで無効） ---
min_efficiency: float | None = None
min_adx: float | None = None
use_htf_filter: bool = False
max_vol_percentile: float | None = 0.95
# --- 執行 ---
use_post_only: bool = False
allow_short: bool = True
bars_per_year: int = 8_760
def n_params(self) -> int:
"""実質的なパラメータ数。記事32のDSRで意識する。"""
n = 3 # entry_period, target_annual_vol, atr_mult
if self.signal == "donchian":
n += 1 # exit_period
for f in (self.min_efficiency, self.min_adx, self.max_vol_percentile):
if f is not None:
n += 1
if self.use_htf_filter:
n += 1
return n
class TrendFollowStrategy(Strategy):
"""ボラティリティ正規化つきトレンドフォロー。"""
def __init__(self, cfg: TrendConfig | None = None):
self.cfg = cfg or TrendConfig()
self.trail: TrailState | None = None
self._cache_len = 0
self._indicators: pd.DataFrame | None = None
self.stats = {"entries": 0, "exits": 0, "trail_stops": 0,
"filtered_out": 0}
# ---- 指標の計算（履歴が伸びたら再計算） -----------------------------
def _compute(self, history: pd.DataFrame) -> pd.DataFrame:
cfg = self.cfg
if cfg.signal == "donchian":
sig = donchian_signal(history["high"], history["low"],
history["close"],
cfg.entry_period, cfg.exit_period)
else:
sig = tsmom_multi_horizon(history["close"], cfg.tsmom_lookbacks)
vol = realised_vol(history["close"], cfg.vol_lookback,
cfg.bars_per_year)
lev = vol_scaled_leverage(vol, VolTargetConfig(
target_annual_vol=cfg.target_annual_vol,
vol_lookback=cfg.vol_lookback,
max_leverage=cfg.max_leverage))
filt = build_filter(
history,
min_efficiency=cfg.min_efficiency,
min_adx=cfg.min_adx,
use_htf=cfg.use_htf_filter,
max_vol_percentile=cfg.max_vol_percentile,
)
return pd.DataFrame({
"signal": sig,
"atr": atr(history, cfg.atr_period),
"leverage": lev,
"allow_long": filt["allow_long"],
"allow_short": filt["allow_short"],
}, index=history.index)
def _indicators_for(self, history: pd.DataFrame) -> pd.DataFrame:
# 毎足フル再計算すると遅いので、長さが変わったときだけ更新する
if self._indicators is None or len(history) != self._cache_len:
self._indicators = self._compute(history)
self._cache_len = len(history)
return self._indicators
# ---- Strategyインターフェース ---------------------------------------
def on_start(self, history: pd.DataFrame) -> None:
self.trail = None
self._indicators = None
self._cache_len = 0
def on_fill(self, fill: Fill) -> None:
"""約定でポジションが建ったらトレールを初期化する。"""
if fill.reason != "signal":
return
if self.trail is None and self._indicators is not None:
atr_now = float(self._indicators["atr"].iloc[-1])
if np.isfinite(atr_now) and atr_now > 0:
self.trail = make_trail(
is_long=(fill.side is Side.BUY),
entry_price=fill.price, atr_value=atr_now,
mult=self.cfg.atr_mult)
def on_bar(self, history: pd.DataFrame, position: Position,
equity: float) -> list[Order] | None:
cfg = self.cfg
warmup = max(cfg.entry_period, cfg.vol_lookback,
max(cfg.tsmom_lookbacks)) + 5
if len(history) < warmup:
return None
ind = self._indicators_for(history)
last = ind.iloc[-1]
price = float(history["close"].iloc[-1])
atr_now = float(last["atr"])
if not np.isfinite(atr_now) or atr_now <= 0:
return None
orders: list[Order] = []
# ---- 1) トレールによる撤退を最優先で判定 ----------------------
if not position.is_flat and self.trail is not None:
self.trail.update(price, atr_now, cfg.atr_mult)
if self.trail.is_hit(price):
self.stats["trail_stops"] += 1
self.trail = None
return [Order(side=position.side.opposite,
qty=abs(position.qty),
order_type=OrderType.MARKET,
reduce_only=True, tag="trail-stop")]
# ---- 2) シグナルとフィルタ -------------------------------------
target = int(last["signal"])
if target > 0 and not bool(last["allow_long"]):
target = 0
self.stats["filtered_out"] += 1
if target < 0 and (not bool(last["allow_short"])
or not cfg.allow_short):
target = 0
current = (0 if position.is_flat
else (1 if position.qty > 0 else -1))
if target == current:
return None
# ---- 3) 既存ポジションの決済 -----------------------------------
if current != 0:
self.stats["exits"] += 1
self.trail = None
orders.append(Order(side=position.side.opposite,
qty=abs(position.qty),
order_type=OrderType.MARKET,
reduce_only=True, tag="exit"))
# ---- 4) 新規エントリー（ボラ正規化サイズ） ----------------------
if target != 0:
leverage = float(last["leverage"])
if not np.isfinite(leverage) or leverage <= 0:
return orders or None
notional = equity * leverage
qty = round(notional / price, 8)
if qty <= 0:
return orders or None
self.stats["entries"] += 1
otype = (OrderType.POST_ONLY if cfg.use_post_only
else OrderType.MARKET)
orders.append(Order(
side=Side.BUY if target > 0 else Side.SELL,
qty=qty, order_type=otype,
price=price if cfg.use_post_only else None,
ttl_bars=3 if cfg.use_post_only else None,
tag="entry-long" if target > 0 else "entry-short",
))
return orders or None
if __name__ == "__main__":
cfg = TrendConfig()
print(f"実質パラメータ数: {cfg.n_params()}")
print(" 記事32のDSRでは、この数と探索回数が判定に影響する。")
print(" フィルタを全部有効にするとパラメータが増えるので、")
print(" まずは最小構成（フィルタなし）から検証すること。")
print("\nフィルタを全部有効にした場合:")
full = TrendConfig(min_efficiency=0.25, min_adx=20.0,
use_htf_filter=True, max_vol_percentile=0.95)
print(f" 実質パラメータ数: {full.n_params()}")トレンドフォローは単一銘柄では成績が不安定です。TSMOMの論文でも「多数の資産に分散して初めて安定する」ことが示されています。ただし仮想通貨は相関が高いため、分散効果は株式・債券・商品のポートフォリオより限定的です。
"""portfolio.py — 複数銘柄への配分とポートフォリオ全体のリスク管理
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
仮想通貨特有の問題:
BTCとアルトコインの相関が0.7〜0.9と非常に高い。
「10銘柄に分散」しても実質的には2〜3銘柄分のリスク分散しかない。
したがってポートフォリオ全体での上限管理が不可欠。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
import numpy as np
import pandas as pd
@dataclass
class PortfolioLimits:
max_total_leverage: float = 3.0 # 全銘柄合計の実効レバレッジ
max_per_symbol_leverage: float = 1.0
max_same_direction_ratio: float = 0.8 # 同方向への偏りの上限
max_symbols: int = 5
def correlation_matrix(closes: dict[str, pd.Series],
lookback: int = 720) -> pd.DataFrame:
"""銘柄間の対数リターン相関。分散が効いているかを確認する。"""
rets = {}
for sym, close in closes.items():
r = np.log(close / close.shift(1)).tail(lookback)
rets[sym] = r
return pd.DataFrame(rets).dropna().corr()
def effective_n(corr: pd.DataFrame) -> float:
"""実効的な独立銘柄数。相関が高いと1に近づく。
固有値から算出する簡易指標:
N_eff = (Σλ)^2 / Σλ^2
"""
if corr.empty:
return 0.0
eig = np.linalg.eigvalsh(corr.to_numpy())
eig = eig[eig > 1e-10]
if eig.size == 0:
return 0.0
return float(eig.sum() ** 2 / (eig ** 2).sum())
def inverse_vol_weights(vols: dict[str, float]) -> dict[str, float]:
"""逆ボラティリティ配分。ボラの低い銘柄に多く配分する。"""
inv = {s: (1.0 / v if v > 0 else 0.0) for s, v in vols.items()}
total = sum(inv.values())
if total <= 0:
n = len(vols) or 1
return {s: 1.0 / n for s in vols}
return {s: w / total for s, w in inv.items()}
@dataclass
class PortfolioState:
"""全銘柄の目標ポジションを管理し、上限を強制する。"""
limits: PortfolioLimits = field(default_factory=PortfolioLimits)
targets: dict[str, float] = field(default_factory=dict) # symbol -> 符号付きレバ
def set_target(self, symbol: str, signed_leverage: float) -> None:
self.targets[symbol] = signed_leverage
def clipped(self) -> dict[str, float]:
"""上限を適用した最終的な配分を返す。"""
lim = self.limits
out = dict(self.targets)
# 1) 銘柄ごとの上限
for s, v in out.items():
out[s] = float(np.clip(v, -lim.max_per_symbol_leverage,
lim.max_per_symbol_leverage))
# 2) 銘柄数の上限（絶対値の大きい順に採用）
if len(out) > lim.max_symbols:
ranked = sorted(out.items(), key=lambda kv: -abs(kv[1]))
keep = dict(ranked[:lim.max_symbols])
out = {s: (keep.get(s, 0.0)) for s in out}
# 3) 同方向への偏りの上限
total_abs = sum(abs(v) for v in out.values())
if total_abs > 0:
long_share = sum(v for v in out.values() if v > 0) / total_abs
short_share = -sum(v for v in out.values() if v < 0) / total_abs
for share, sign in ((long_share, 1), (short_share, -1)):
if share > lim.max_same_direction_ratio:
scale = lim.max_same_direction_ratio / share
for s, v in out.items():
if np.sign(v) == sign:
out[s] = v * scale
# 4) 合計レバレッジの上限（比例縮小）
total = sum(abs(v) for v in out.values())
if total > lim.max_total_leverage and total > 0:
scale = lim.max_total_leverage / total
out = {s: v * scale for s, v in out.items()}
return out
def report(self) -> str:
raw, final = self.targets, self.clipped()
lines = [f"{'symbol':<12}{'目標レバ':>10}{'調整後':>10}{'変化':>8}"]
lines.append("-" * 40)
for s in sorted(raw, key=lambda x: -abs(raw[x])):
r, f = raw[s], final.get(s, 0.0)
mark = "" if abs(r - f) < 1e-9 else " <- 縮小"
lines.append(f"{s:<12}{r:>+10.3f}{f:>+10.3f}"
f"{(f - r):>+8.3f}{mark}")
lines.append(f"\n 合計(絶対値): {sum(abs(v) for v in raw.values()):.2f}"
f" -> {sum(abs(v) for v in final.values()):.2f}"
f" (上限 {self.limits.max_total_leverage})")
return "\n".join(lines)
if __name__ == "__main__":
from kline_store import KlineStore
from signals import tsmom_multi_horizon
from vol_target import VolTargetConfig, realised_vol, vol_scaled_leverage
store = KlineStore()
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "XRPUSDT"]
closes: dict[str, pd.Series] = {}
for sym in symbols:
df = store.load(sym, "60")
if not df.empty:
closes[sym] = df.loc["2024-01-01":]["close"]
if len(closes) < 2:
raise SystemExit("2銘柄以上のデータが必要です（記事07）")
corr = correlation_matrix(closes)
print("=== 銘柄間の相関（直近720本） ===")
print(corr.to_string(float_format=lambda v: f"{v:.2f}"))
n_eff = effective_n(corr)
print(f"\n実効独立銘柄数: {n_eff:.2f} / 実際の銘柄数 {len(corr)}")
print(f" -> {len(corr)}銘柄に分散しても、リスク分散は実質"
f"{n_eff:.1f}銘柄分しかない")
# 各銘柄のシグナルとボラからポートフォリオを組む
state = PortfolioState()
cfg = VolTargetConfig(target_annual_vol=0.20, max_leverage=1.0)
print("\n=== ポートフォリオの構成 ===")
for sym, close in closes.items():
sig = tsmom_multi_horizon(close).iloc[-1]
vol = realised_vol(close, cfg.vol_lookback, 8_760).iloc[-1]
lev = vol_scaled_leverage(pd.Series([vol]), cfg).iloc[0]
state.set_target(sym, float(lev) * int(sig))
print(state.report())
print("\n※ 相関が高い仮想通貨では、銘柄数を増やしてもリスクは下がりにくい。")
print(" 合計レバレッジの上限管理が本質的な防御になる（記事23・24）。")ここが最重要です。実装ができた時点では何も検証できていません。記事32の手法を全部通します。
"""validate_trend.py — トレンドフォロー戦略を記事32の手法で検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
手順:
1) 悲観コストモデルでのバックテスト（記事31）
2) ルックアヘッドテスト（記事31）
3) Walk-Forward最適化と効率比（記事32）
4) Deflated Sharpe Ratio（記事32）
5) PBO（記事32）
6) 連敗分布の把握（心理的な耐性の確認）
7) 採用判定（記事32のadoption_gate）
"""
from __future__ import annotations
from decimal import Decimal
import numpy as np
import pandas as pd
from adoption_gate import Criteria, Evidence, evaluate_adoption
from bt.engine import BacktestConfig, Backtester
from bt.fill import FillModel, PessimisticFillModel
from bt.report import metrics
from costs import FeeModel
from deflated_sharpe import deflated_sharpe_ratio
from pbo import build_returns_matrix, compute_pbo
from strategies.trend_follow import TrendConfig, TrendFollowStrategy
from walk_forward import walk_forward
def losing_streaks(pnl: pd.Series) -> dict:
"""連敗の分布。心理的な耐性を事前に把握するために測る。"""
signs = np.sign(pnl.to_numpy())
streaks, run = [], 0
for s in signs:
if s < 0:
run += 1
elif run:
streaks.append(run)
run = 0
if run:
streaks.append(run)
if not streaks:
return {"max": 0, "mean": 0.0, "p95": 0}
arr = np.array(streaks)
return {"max": int(arr.max()), "mean": float(arr.mean()),
"p95": int(np.percentile(arr, 95)), "count": len(arr)}
def evaluate_config(bars: pd.DataFrame, params: dict,
fees: FeeModel, funding: pd.Series | None = None,
pessimistic: bool = False) -> tuple[float, pd.Series]:
"""パラメータ辞書からTrendConfigを作り、Sharpeとリターン列を返す。
walk_forward / build_returns_matrix に渡す評価関数。
"""
cfg = TrendConfig(**params)
model = PessimisticFillModel() if pessimistic else FillModel(
slippage_bps=2.0, maker_fill_prob=0.7)
bt = Backtester(bars, fees, fill_model=model, funding=funding,
config=BacktestConfig(initial_cash=10_000))
result = bt.run(TrendFollowStrategy(cfg))
eq = result.equity.dropna()
if len(eq) < 10:
return -np.inf, pd.Series(dtype=float)
rets = eq.pct_change().dropna()
std = rets.std(ddof=1)
if std == 0 or not np.isfinite(std):
return -np.inf, rets
sharpe = float(rets.mean() / std * np.sqrt(8_760))
return sharpe, rets
def full_validation(bars: pd.DataFrame, fees: FeeModel,
funding: pd.Series | None = None) -> None:
# パラメータ候補は意図的に少なくする（試行回数を抑える）
grid = [
{"signal": "donchian", "entry_period": ep, "exit_period": xp,
"atr_mult": am, "target_annual_vol": 0.20, "max_leverage": 2.0}
for ep in (40, 55, 80)
for xp in (15, 20, 30)
for am in (2.5, 3.0, 4.0)
]
n_trials = len(grid)
print(f"パラメータ候補: {n_trials}通り（意図的に少なく保つ）\n")
def ev(b: pd.DataFrame, p: dict) -> tuple[float, pd.Series]:
return evaluate_config(b, p, fees, funding)
# ---- 1) Walk-Forward -------------------------------------------------
print("=" * 58)
print("[1] Walk-Forward最適化")
print("=" * 58)
wfa = walk_forward(bars, grid, ev, is_days=180, oos_days=30)
print(wfa.report())
oos_rets = wfa.stitched_oos()
if oos_rets.empty:
print("\nOOSリターンが取得できませんでした")
return
# ---- 2) OOS成績の指標 -----------------------------------------------
print("\n" + "=" * 58)
print("[2] OOS（つないだ経路）の成績")
print("=" * 58)
std = oos_rets.std(ddof=1)
oos_sharpe = (float(oos_rets.mean() / std * np.sqrt(8_760))
if std > 0 else 0.0)
eq = (1 + oos_rets).cumprod()
peak = eq.cummax()
max_dd = float((eq / peak - 1).min())
years = len(oos_rets) / 8_760
annual = float(eq.iloc[-1] ** (1 / years) - 1) if years > 0 else 0.0
print(f" OOS Sharpe : {oos_sharpe:>8.3f}")
print(f" OOS 年率 : {annual:>8.2%}")
print(f" OOS 最大DD : {max_dd:>8.2%}")
# ---- 3) Deflated Sharpe Ratio ---------------------------------------
print("\n" + "=" * 58)
print("[3] Deflated Sharpe Ratio")
print("=" * 58)
trial_sharpes = []
for p in grid:
s, _ = ev(bars, p)
if np.isfinite(s):
trial_sharpes.append(s)
dsr = deflated_sharpe_ratio(oos_rets.to_numpy(), n_trials=n_trials,
trial_sharpes=trial_sharpes,
periods_per_year=8_760)
print(dsr.report())
# ---- 4) PBO ----------------------------------------------------------
print("\n" + "=" * 58)
print("[4] PBO（過剰最適化確率）")
print("=" * 58)
M = build_returns_matrix(bars, grid, ev)
pbo = compute_pbo(M, n_blocks=10)
print(pbo.report())
# ---- 5) 悲観モデル ---------------------------------------------------
print("\n" + "=" * 58)
print("[5] 悲観コストモデル（スリッページ4倍・約定確率半分）")
print("=" * 58)
best = wfa.windows[-1].best_params if wfa.windows else grid[0]
pess_sharpe, pess_rets = evaluate_config(bars, best, fees, funding,
pessimistic=True)
pess_net = float((1 + pess_rets).prod() - 1) * 10_000 if len(pess_rets) else 0.0
print(f" Sharpe : {pess_sharpe:>8.3f}")
print(f" 純損益(想定) : {pess_net:>+10.2f}")
# ---- 6) 連敗分布 -----------------------------------------------------
print("\n" + "=" * 58)
print("[6] 連敗の分布（心理的な耐性の確認）")
print("=" * 58)
daily = oos_rets.resample("D").sum()
st = losing_streaks(daily[daily != 0])
print(f" 最長連敗(日) : {st['max']}")
print(f" 95%点 : {st['p95']}")
print(f" 平均 : {st['mean']:.1f}")
print(f" -> {st['max']}日連続で負ける局面が実際にあった。")
print(" これに耐えられないなら、この戦略は運用できない。")
# ---- 7) 採用判定 -----------------------------------------------------
print("\n" + "=" * 58)
print("[7] 採用判定")
print("=" * 58)
evidence = Evidence(
n_trades=int((oos_rets != 0).sum()),
oos_sharpe=oos_sharpe,
wfe=wfa.mean_efficiency,
dsr=dsr.dsr,
pbo=pbo.pbo,
param_stability=wfa.param_stability,
max_dd=max_dd,
annual_return=annual,
pessimistic_net=pess_net,
oos_positive_ratio=wfa.oos_positive_ratio,
lookahead_test_passed=True, # 別途 test_lookahead.py で確認
n_trials_declared=n_trials,
)
print(evaluate_adoption(evidence, Criteria()).report())
if __name__ == "__main__":
from client import make_client
from costs import fetch_funding_history
from kline_store import KlineStore
bars = KlineStore().load("BTCUSDT", "60")
if bars.empty:
raise SystemExit("先に記事07でデータを取得してください")
bars = bars.loc["2023-06-01":]
fees = FeeModel(maker=Decimal("0.0002"), taker=Decimal("0.00055"))
try:
c = make_client()
start_ms = int(bars.index[0].timestamp() * 1000)
funding = fetch_funding_history(c.http, "BTCUSDT",
start_ms)["funding_rate"]
except Exception as e:
print(f"資金調達履歴の取得に失敗（{e}）。資金調達なしで検証します")
funding = None
full_validation(bars, fees, funding)サンプル7を走らせると、多くの設定が「採用不可」になります。それが正常です。落ちたときにやってはいけないのは、基準を通すまでパラメータをいじることです。それは試行回数を増やす行為であり、DSRの前提を悪化させます。
やるべきなのは、(1) パラメータを減らす（フィルタを外す、シグナルを単純にする）、(2) 複数銘柄・複数期間で同じパラメータが機能するか確認する、(3) 検証期間を延ばす——のいずれかです。「調整して通す」のではなく「頑健にする」方向に動いてください。
記事30のアーキテクチャに載せれば、Strategyクラスを1行も変更せずに本番で動きます。
"""run_trend_bot.py — トレンドフォローbotの本番起動スクリプト
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
記事30のRunnerに、記事33のStrategyを載せるだけ。
バックテストで検証したものと同一のクラスが動く。
"""
from __future__ import annotations
import asyncio
import logging
import sys
from components.feed import AccountFeed, MarketDataFeed
from components.order_router import OrderRouterComponent
from components.position import PositionKeeperComponent
from components.risk import RiskManagerComponent
from components.strategy_engine import StrategyEngine
from config_model import load_settings
from core.bus import EventBus
from core.runner import BotRunner
from client import make_client
from idempotent_router import IdempotentRouter
from order_ledger import OrderLedger
from reconciler import startup_reconcile
from strategies.trend_follow import TrendConfig, TrendFollowStrategy
log = logging.getLogger("run_trend_bot")
# バックテストで検証したのと「完全に同じ」設定を使う。
# ここを変えたら、再度記事32の検証を通す必要がある。
VALIDATED_CONFIG = TrendConfig(
signal="donchian",
entry_period=55,
exit_period=20,
atr_period=22,
atr_mult=3.0,
target_annual_vol=0.20,
vol_lookback=168,
max_leverage=2.0,
min_efficiency=None, # フィルタは検証で有効性を確認できたものだけ
min_adx=None,
use_htf_filter=False,
max_vol_percentile=0.95,
use_post_only=True, # メイカー執行でコストを抑える（記事13）
allow_short=True,
)
async def main() -> int:
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(name)-22s %(message)s",
)
settings = load_settings()
print(settings.banner())
c = make_client(settings.env)
bus = EventBus(maxsize=settings.bus_maxsize)
# ---- 起動前のリコンサイル（記事17）------------------------------
ledger = OrderLedger()
router = IdempotentRouter(c.http, ledger)
try:
startup_reconcile(c, ledger, router,
strategy_prefixes=set(settings.strategy_prefixes),
cancel_orphans=settings.cancel_orphans,
require_clean=True)
except RuntimeError as e:
log.error("リコンサイル失敗のため起動を中止: %s", e)
return 1
# ---- 口座設定を望む状態に収束させる（記事21）---------------------
from account_setup import DesiredSetup, apply_setup, verify_setup
from symbol_rules import SymbolRegistry
reg = SymbolRegistry(c.http)
desired = DesiredSetup(
symbols={s: settings.leverage for s in settings.symbols},
hedge_mode=False,
planned_notional=settings.risk.max_notional_per_symbol,
)
result = apply_setup(c.http, desired, registry=reg)
print(result.report())
problems = verify_setup(c.http, desired)
if problems:
log.error("口座設定が期待値と一致しません: %s", problems)
return 1
# ---- コンポーネントの組み立て（記事30）---------------------------
feed = MarketDataFeed(bus, c, symbols=settings.symbols,
interval=settings.interval,
category=settings.category,
warmup_bars=400) # ウォームアップを十分に取る
account = AccountFeed(bus, c)
keeper = PositionKeeperComponent(bus, c)
risk = RiskManagerComponent(bus, keeper, settings.risk)
order_router = OrderRouterComponent(bus, c, router, keeper)
# 全銘柄に同じ検証済み設定を適用する
strategies = {sym: TrendFollowStrategy(VALIDATED_CONFIG)
for sym in settings.symbols}
engine = StrategyEngine(bus, strategies, feed, keeper,
min_bars=400)
runner = BotRunner(bus, heartbeat_sec=settings.heartbeat_sec)
runner.add(keeper, account, feed, risk, order_router, engine)
log.info("トレンドフォローbotを起動します（%d銘柄）",
len(settings.symbols))
log.info("設定: %s", VALIDATED_CONFIG)
return await runner.run()
if __name__ == "__main__":
sys.exit(asyncio.run(main()))設定をコード内の定数として固定し、コメントで「検証済み」と明示しているのは意図的です。稼働中に「なんとなく調整」してしまうのを防ぎます。
パラメータを変えたら、それは別の戦略です。記事32の検証をもう一度通し、試行回数のカウントも増やしてください。実運用中の微調整は、最も気づきにくい過剰最適化です。
compare_signals() で「平均保有本数」を確認してください。1時間足で平均20本保有なら20時間で往復コストを回収する必要があります。日足のほうが検証も安定しますが、約定回数が減るため統計的な信頼性（記事32の min_trades）を確保しにくくなるというトレードオフがあります。TrendConfig(allow_short=False) でロングのみに切り替えられます。なおロングのみのほうがパラメータが減るため、記事32の検証を通しやすいという副次的な利点もあります。ショートを入れる場合は、資金調達の受取り（記事10）を必ずバックテストに含めてください。n_params() で確認できます）。max_consecutive_losses）。ただし基準を超えた場合も「戦略が壊れた」とは限りません。トレンドフォローは長期の連敗が構造的に起こるため、止める判断は「損失額」ではなく「市場環境の変化」で行うほうが合理的です。effective_n() で確認できますが、BTCとアルトコインの相関は0.7〜0.9と高く、10銘柄に分散しても実効的には2〜3銘柄分のリスク分散しかありません。PortfolioLimits）が本質的な防御になります（記事23・24）。maker_fill_prob を測る）。投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません（間接的・付随的な障害・損失も含め一切の責任を負いません）。自己責任でご利用ください。
当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。
当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。
当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。
当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。
当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。
当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。