🔹戦略概要（VwapMacdTrend）
VWAPとMACDヒストグラムを組み合わせた短期ショート戦略。
下方向のトレンドが明確な場面だけに入ることで、寄り天・下落トレンドに乗ることを狙う。
🔸取引ルールまとめ
1. 時間フィルタ
- 取引許可時間：9:05～14:30
- 14:55 で全ポジションをクローズ（引け前フラット）
2. エントリー条件（ショート限定）
- 現在価格 ＜ VWAP
- VWAPが 前バーよりも下向き（vwap_slope < 0）
- MACDヒストグラム < 0 かつ悪化中（hist < hist_prev * 1.05）
- VWAPからの価格乖離が ±0.5％以内
- ATRが十分に大きい（ATR / price > 0.1％）
- 出来高が最近20本の中央値の 30%以上
- すべて満たしたときにショート発注
3. 損切り・利確設定
- 損切り(SL)
直近5本の高値と「ATR×倍率」のうち厳しい方を採用し、
さらに +ε（例：1円）を足して必ず価格より上に置く。 - 利確(TP)
エントリー価格から下方向に 1.2R（Rはリスク）−0.5円
→「リスクリワード比 1:1.2」で設定 - ロット（size）
口座資金の0.3%をリスクに割り当てて自動算出。size = equity * risk_per_trade / risk
4. ポジション管理
- 同一バーで複数発注しない（
last_entry_iで1バー1発注制御） - すでにポジションがある場合は：
- 経過時間が20分を超えて +0.5R未満 → 撤退
- MACDヒストグラムが改善 or 価格がVWAPを上抜け → 撤退
5. 実行環境の設定
- 約定タイミング：同バー終値（trade_on_close=True）
- 同時注文を禁止（exclusive_orders=True）
- 取引コスト：0（手数料ゼロ仮定）
- データ：yfinanceから1分足・7日間取得
6. デバッグ・記録
- 各条件でブロックされた回数を
block辞書でカウント
→time,dist,vol,atr,cond,ok,pos - 取引履歴（_trades）をCSV出力
7. 改善後の特徴
- 約定が安定（ε補正とexclusive_ordersの導入）
- 実際の二重発注はなく、ログ出力が重複していただけ
- 条件を緩めたことでエントリーが発生しやすくなり、
5分足や7日分でも複数トレードが成立
そしてコーディングしてもらった結果がこれ。BackTest03-toyota-short.py
ロングで勝てないのは、この期間のトヨタはギャップアップで寄り付いた後は下がるからなのかと考えて、ショートなら勝てるかと思ったけど、ロングで勝てない戦略ではショートでも勝てないことが分かった。
/
import yfinance as yf
import pandas as pd
import pandas_ta as ta
from backtesting import Backtest, Strategy
from datetime import time
import pytz
ALLOW_LONG = False
ALLOW_SHORT = True
brand_names = {
"1570.T": "Nikkei NF",
"6526.T": "Sociotech",
"7203.T": "Toyota",
"6857.T": "Advantest",
"6146.T": "Disco",
"4062.T": "Ibiden",
"6920.T": "LaserTech",
"5803.T": "Fujikura",
"8035.T": "Tokyo Electron",
"6758.T": "SONY group",
"9984.T": "Softbank group",
"AAPL": "Apple Inc.",
}
# ===== 1) データ取得（列を最初からフラットに） =====
TICKER = '7203.T'
INTERVAL = "1m"
PERIOD = "7d"
df = yf.download(
TICKER, interval=INTERVAL, period=PERIOD,
auto_adjust=False, group_by="column" # ← これで ['Open','High',...] の一次元列
).dropna()
# --- ダウンロード直後に追加（列の正規化） ---
import pandas as pd
# 1) MultiIndex → 1段化（安全）
if isinstance(df.columns, pd.MultiIndex):
df.columns = ['_'.join([str(x) for x in c if x]).strip() for c in df.columns]
# 2) 1570.T の接尾辞付き列を素名にコピー
def pick_base(name: str):
# 完全一致 or "name_..." を優先的に拾う
cands = [c for c in df.columns if c == name] \
or [c for c in df.columns if c.startswith(name + '_')]
return cands[0] if cands else None
base_map = {}
for name in ['Open', 'High', 'Low', 'Close', 'Volume']:
src = pick_base(name)
if src is None:
raise KeyError(f"{name} 列が見つかりません。現在の列: {list(df.columns)}")
df[name] = df[src] # 素名で上書き作成
# 3) 余分なNaN除去（お好みで）
df.dropna(subset=['Open','High','Low','Close','Volume'], inplace=True)
# --- VWAP（安定版） ---
def intraday_vwap(df):
tp = (df['High'] + df['Low'] + df['Close']) / 3
idx = df.index
if getattr(idx, "tz", None) is None:
idx = idx.tz_localize("Asia/Tokyo")
day_key = pd.Series(idx.tz_convert("Asia/Tokyo").date, index=df.index)
cum_vol = df['Volume'].groupby(day_key).cumsum()
cum_vp = (tp * df['Volume']).groupby(day_key).cumsum()
return (cum_vp / cum_vol.where(cum_vol != 0, pd.NA)).astype('float64')
df['VWAP'] = intraday_vwap(df)
# --- MACD（確実に作る版） ---
import pandas_ta as ta
macd = ta.macd(df['Close'], fast=8, slow=21, signal=9, append=True)
if isinstance(macd, pd.DataFrame):
# 列名を統一
if isinstance(macd.columns, pd.MultiIndex):
macd.columns = ['_'.join([str(x) for x in c if x]).strip() for c in macd.columns]
macd.rename(columns={
'MACD_8_21_9': 'MACD',
'MACDs_8_21_9': 'SIGNAL',
'MACDh_8_21_9': 'HIST'
}, inplace=True)
df = df.join(macd)
else:
# もし None を返す版なら append=True で追加
df.ta.macd(close='Close', fast=12, slow=26, signal=9, append=True)
# 追加された列を拾ってリネーム
rename_map = {}
for c in list(df.columns):
s = str(c)
if 'MACD_12_26_9' in s and 'MACDs' not in s and 'MACDh' not in s:
rename_map[c] = 'MACD'
if 'MACDs_12_26_9' in s:
rename_map[c] = 'SIGNAL'
if 'MACDh_12_26_9' in s:
rename_map[c] = 'HIST'
if rename_map:
df.rename(columns=rename_map, inplace=True)
# 最終チェック
for col in ['MACD','SIGNAL','HIST']:
if col not in df.columns:
print("現在の列:", list(df.columns))
raise KeyError(f"MACD列 {col} が見つかりません")
# --- Backtestに渡す列だけに絞る ---
need = ['Open','High','Low','Close','Volume','VWAP','HIST']
df = df[need].dropna()
JST = pytz.timezone("Asia/Tokyo")
FLAT_TIME = time(14, 55) # 14:55にフラットにする例
# ===== 5) 戦略 ===== TOYOTA向け
from datetime import time, timedelta
import pytz
JST = pytz.timezone("Asia/Tokyo")
# === パラメータ（調整しやすい所） ===
OPEN_OK_FROM = time(9, 5)
NO_NEW_AFTER = time(14, 30)
FLAT_TIME = time(14, 55)
VWAP_DIST_MAX = 0.005 # 0.2%
VOL_MED_WIN = 20
VOL_FLOOR_MULT = 0.3
ATR_PCT_MIN = 0.0010 # 0.35%
TIME_STOP_MIN = 20 # 分
R_MULT_TP = 1.2 # 全利確のR倍率（1.5〜2.0で試す）
class VwapMacdTrend(Strategy):
sl_ticks = 15 # 最小損切り幅
atr_mult = 1.0
risk_per_trade = 0.003 # コスト抑制で一旦0.5%
block = None # デバッグカウンタ
def init(self):
# 指標
self.last_entry_i = -1
# 実行のたびにリセット
type(self).block = dict(time=0, dist=0, vol=0, atr=0, cond=0, ok=0, pos=0)
self.atr = self.I(ta.atr, pd.Series(self.data.High),
pd.Series(self.data.Low),
pd.Series(self.data.Close), 14)
# 出来高中央値
self.vol_med = self.I(lambda v: pd.Series(v).rolling(VOL_MED_WIN).median().bfill(),
self.data.Volume)
def _jst_time(self):
return self.data.index[-1].to_pydatetime().astimezone(JST).time()
def next(self):
# --- 引け前はフラット ---
if self._jst_time() >= FLAT_TIME and self.position:
self.position.close(); return
price = self.data.Close[-1]
vwap = self.data.df['VWAP'].iloc[-1]
vwap_prev = self.data.df['VWAP'].iloc[-2]
hist = self.data.df['HIST'].iloc[-1]
hist_prev = self.data.df['HIST'].iloc[-2]
vwap_slope = vwap - vwap_prev
# --- 既存ポジ管理（タイムストップ & 失速/割れ） ---
if self.position:
trade = self.trades[-1]
dur = self.data.index[-1] - self.data.index[trade.entry_bar]
# エントリー時の想定リスク（R）を復元
#r = max(trade.entry_price - (trade.sl or trade.entry_price - self.sl_ticks), 1e-6)
r = max((trade.sl or trade.entry_price + self.sl_ticks) - trade.entry_price, 1e-6)
# タイムストップ：45分で +0.5R 未満なら撤退
upnl_r = (trade.entry_price - price) / r
if dur >= timedelta(minutes=TIME_STOP_MIN) and upnl_r < 0.5:
self.position.close(); return
# 失速/割れ
if (hist > hist_prev) or (price > vwap):
self.position.close()
return
# ── ここから新規エントリーの判定（ショート） ──
# 時間帯
t = self._jst_time()
if not (OPEN_OK_FROM <= t < NO_NEW_AFTER):
type(self).block['time'] += 1; return
# 飛びつき防止：VWAPからの距離
if abs(price - vwap)/max(vwap,1) > VWAP_DIST_MAX:
type(self).block['dist'] += 1; return
# ボラ不足
if (self.atr[-1] / max(price,1)) < ATR_PCT_MIN:
type(self).block['atr'] += 1; return
# 出来高
if self.data.Volume[-1] < VOL_FLOOR_MULT * max(self.vol_med[-1], 1):
type(self).block['vol'] += 1; return
# ショートの“二段目”：価格＜VWAP、VWAP↓、ヒスト<0 かつ悪化（減少）
vwap_slope = vwap - vwap_prev
hist_down = (hist <= hist_prev * 1.05) and (hist < 0) # 少し緩め（悪化）
short_ok = (price < vwap) and (vwap_slope < 0) and hist_down
if not short_ok:
type(self).block['cond'] += 1; return
type(self).block['ok'] += 1
# 損切り（ショート）：直近高値とATR基準の厳しい方
atr = max(float(self.atr[-1]), 1e-6)
recent_high = max(self.data.High[-5:])
sl_price = max(recent_high, price + max(self.sl_ticks, atr * self.atr_mult)) # ← 価格より上
risk = max(sl_price - price, 1e-6) # ← 上側がリスク
tp = price - R_MULT_TP * risk # ← 目標は下
size = max(int((self.equity * self.risk_per_trade) / risk), 1) # ← 最低1
tp = price - R_MULT_TP * risk # 目標利確は下
# 価格・ATRなど
price = self.data.Close[-1]
atr = max(float(self.atr[-1]), 1e-6)
# 1) ショートのSLは「上側」：直近高値とATR基準の厳しい方に ε を足す
recent_high = max(self.data.High[-5:])
raw_sl = max(recent_high, price + max(self.sl_ticks, atr * self.atr_mult))
EPS = 1.0
sl_price = raw_sl + EPS # ← ε（0.5円など）で “必ず価格より上” を保証
# 2) リスク（R）とサイズ
risk = max(sl_price - price, 1e-6) # ← ショートは上側がR
size = max(int((self.equity * self.risk_per_trade) / risk), 1)
# 3) TPは「下側」：R倍の距離に ε を引く
tp = price - R_MULT_TP * risk - 0.5 # ← εで “必ず価格より下” を保証
# 4) 約定
# 発注
if self.last_entry_i == len(self.data) - 1:
return # そのバーは既に発注済み
# ↓ ここで 1 回だけ発注する
oid = self.sell(size=size, sl=sl_price, tp=tp)
self.last_entry_i = len(self.data) - 1
type(self).block['pos'] += 1
print("TRY SELL",
"t=", self._jst_time(),
"price=", price, "vwap=", vwap,
"sl=", sl_price, "tp=", tp, "risk=", risk)
# ===== 6) 実行 =====
bt = Backtest(
df, VwapMacdTrend,
cash=1_000_000,
commission=0.0,
trade_on_close=True, # ← 同バーの終値で約定
# shorting=True,
exclusive_orders=True # ← 同時に1注文だけ通す
)
stats = bt.run()
print(stats)
print("BLOCK:", VwapMacdTrend.block)
# 取引明細を確認
print(stats['_trades'].head())
# CSVに保存
stats['_trades'].to_csv('trades.csv', index=False)
#print(bt.run()._strategy.block)
bt.plot(open_browser=True)
コメント