feat: add sim_45m40.py and update STRATEGY.md with 40min backtest results
This commit is contained in:
joungmin
296
sim_45m40.py
Normal file
296
sim_45m40.py
Normal file
@@ -0,0 +1,296 @@
|
||||
"""45일 복리 KRW 시뮬레이션 — 40분봉.
|
||||
|
||||
sim10m_cache.pkl(10분봉)을 40분봉으로 리샘플링 후
|
||||
sim_365.py 와 동일한 복리·WF·MAX_POSITIONS 로직 적용.
|
||||
"""
|
||||
|
||||
import pickle
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv(dotenv_path=Path(__file__).parent / ".env")
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
# ── 파라미터 ─────────────────────────────────────────────
|
||||
CACHE_FILE = Path("sim10m_cache.pkl")
|
||||
TOP30_FILE = Path("top30_tickers.pkl")
|
||||
TOP_N = 20
|
||||
|
||||
BUDGET = 15_000_000
|
||||
MIN_BUDGET = BUDGET * 3 // 10
|
||||
MAX_POS = 3
|
||||
|
||||
FEE = 0.0005
|
||||
TIME_STOP_MIN_PCT = 3.0
|
||||
ATR_MULT = 1.5
|
||||
ATR_MIN = 0.010
|
||||
ATR_MAX = 0.020
|
||||
|
||||
VOL_MULT = 2.0
|
||||
QUIET_PCT = 2.0
|
||||
THRESH = 4.8
|
||||
|
||||
# 40분봉 기준 시간 파라미터 → 봉수 환산 (60/40 = 1.5봉/h)
|
||||
LOCAL_VOL_N = 7 # 5h × 1.5
|
||||
QUIET_N = 3 # 2h × 1.5
|
||||
SIGNAL_TO_N = 12 # 8h × 1.5
|
||||
ATR_N = 7 # 5h × 1.5
|
||||
TS_N = 12 # 8h × 1.5
|
||||
|
||||
WF_WINDOW = 2
|
||||
WF_MIN_WIN_RATE = 0.01
|
||||
WF_SHADOW_WINS = 2
|
||||
|
||||
|
||||
# ── 리샘플링 ─────────────────────────────────────────────
|
||||
def resample_40m(df: pd.DataFrame) -> pd.DataFrame:
|
||||
return (
|
||||
df.resample("40min")
|
||||
.agg({"open": "first", "high": "max", "low": "min",
|
||||
"close": "last", "volume": "sum"})
|
||||
.dropna(subset=["close"])
|
||||
)
|
||||
|
||||
|
||||
# ── ATR ──────────────────────────────────────────────────
|
||||
def calc_atr(df: pd.DataFrame, buy_idx: int) -> float:
|
||||
sub = df.iloc[max(0, buy_idx - ATR_N - 1):buy_idx]
|
||||
if len(sub) < 3:
|
||||
return ATR_MIN
|
||||
try:
|
||||
avg = ((sub["high"] - sub["low"]) / sub["low"]).iloc[-ATR_N:].mean()
|
||||
return float(max(ATR_MIN, min(ATR_MAX, avg * ATR_MULT)))
|
||||
except Exception:
|
||||
return ATR_MIN
|
||||
|
||||
|
||||
# ── 포지션 시뮬 ──────────────────────────────────────────
|
||||
def simulate_pos(df: pd.DataFrame, buy_idx: int,
|
||||
buy_price: float, stop_pct: float):
|
||||
peak = buy_price
|
||||
for i in range(buy_idx + 1, len(df)):
|
||||
row = df.iloc[i]
|
||||
ts = df.index[i]
|
||||
if row["high"] > peak:
|
||||
peak = row["high"]
|
||||
if row["low"] <= peak * (1 - stop_pct):
|
||||
sp = peak * (1 - stop_pct)
|
||||
pnl = (sp * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
|
||||
return pnl > 0, ts, pnl
|
||||
pnl_now = (row["close"] - buy_price) / buy_price * 100
|
||||
if (i - buy_idx) >= TS_N and pnl_now < TIME_STOP_MIN_PCT:
|
||||
pnl = (row["close"] * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
|
||||
return pnl > 0, ts, pnl
|
||||
last = df.iloc[-1]["close"]
|
||||
pnl = (last * (1 - FEE) - buy_price * (1 + FEE)) / (buy_price * (1 + FEE)) * 100
|
||||
return pnl > 0, df.index[-1], pnl
|
||||
|
||||
|
||||
# ── vol-lead 전략 ─────────────────────────────────────────
|
||||
def run_vol_lead(df: pd.DataFrame, ticker: str) -> list:
|
||||
trades = []
|
||||
sig_i = sig_p = None
|
||||
in_pos = False
|
||||
buy_idx = buy_price = stop_pct = None
|
||||
i = max(LOCAL_VOL_N + 2, QUIET_N + 1)
|
||||
|
||||
while i < len(df):
|
||||
if in_pos:
|
||||
is_win, sdt, pnl = simulate_pos(df, buy_idx, buy_price, stop_pct)
|
||||
next_i = next((j for j in range(i, len(df)) if df.index[j] > sdt), len(df))
|
||||
trades.append((is_win, pnl, df.index[buy_idx], sdt, ticker))
|
||||
in_pos = False
|
||||
sig_i = sig_p = None
|
||||
i = next_i
|
||||
continue
|
||||
|
||||
close = df.iloc[i]["close"]
|
||||
vol_p = df.iloc[i - 1]["volume"]
|
||||
vol_avg = df.iloc[i - LOCAL_VOL_N - 1:i - 1]["volume"].mean()
|
||||
vol_r = vol_p / vol_avg if vol_avg > 0 else 0
|
||||
|
||||
close_qh = df.iloc[i - QUIET_N]["close"]
|
||||
chg_qh = abs(close - close_qh) / close_qh * 100
|
||||
quiet = chg_qh < QUIET_PCT
|
||||
spike = vol_r >= VOL_MULT
|
||||
|
||||
if quiet and spike:
|
||||
if sig_i is None:
|
||||
sig_i, sig_p = i, close
|
||||
else:
|
||||
if sig_i is not None and close < sig_p:
|
||||
sig_i = sig_p = None
|
||||
if sig_i is not None and (i - sig_i) > SIGNAL_TO_N:
|
||||
sig_i = sig_p = None
|
||||
if sig_i is not None and (close - sig_p) / sig_p * 100 >= THRESH:
|
||||
in_pos = True
|
||||
buy_idx = i
|
||||
buy_price = close
|
||||
stop_pct = calc_atr(df, i)
|
||||
sig_i = sig_p = None
|
||||
i += 1
|
||||
return trades
|
||||
|
||||
|
||||
# ── WF 필터 ──────────────────────────────────────────────
|
||||
def apply_wf(trades: list) -> tuple:
|
||||
history = []
|
||||
shadow_streak = 0
|
||||
blocked = False
|
||||
accepted = []
|
||||
blocked_cnt = 0
|
||||
|
||||
for trade in trades:
|
||||
is_win = int(trade[0])
|
||||
if not blocked:
|
||||
accepted.append(trade)
|
||||
history.append(is_win)
|
||||
if len(history) >= WF_WINDOW:
|
||||
wr = sum(history[-WF_WINDOW:]) / WF_WINDOW
|
||||
if wr < WF_MIN_WIN_RATE:
|
||||
blocked = True
|
||||
shadow_streak = 0
|
||||
else:
|
||||
blocked_cnt += 1
|
||||
if is_win:
|
||||
shadow_streak += 1
|
||||
if shadow_streak >= WF_SHADOW_WINS:
|
||||
blocked = False
|
||||
history = []
|
||||
shadow_streak = 0
|
||||
else:
|
||||
shadow_streak = 0
|
||||
|
||||
return accepted, blocked_cnt
|
||||
|
||||
|
||||
# ── MAX_POSITIONS 필터 ────────────────────────────────────
|
||||
def apply_max_positions(all_trades: list) -> tuple:
|
||||
open_exits, accepted, skipped = [], [], []
|
||||
for trade in all_trades:
|
||||
buy_dt, sell_dt = trade[2], trade[3]
|
||||
open_exits = [s for s in open_exits if s > buy_dt]
|
||||
if len(open_exits) < MAX_POS:
|
||||
open_exits.append(sell_dt)
|
||||
accepted.append(trade)
|
||||
else:
|
||||
skipped.append(trade)
|
||||
return accepted, skipped
|
||||
|
||||
|
||||
# ── 복리 시뮬 ────────────────────────────────────────────
|
||||
def simulate(accepted: list) -> dict:
|
||||
portfolio = float(BUDGET)
|
||||
total_krw = 0.0
|
||||
monthly = {}
|
||||
trade_log = []
|
||||
|
||||
for is_win, pnl, buy_dt, sell_dt, ticker in accepted:
|
||||
pos_size = max(portfolio, MIN_BUDGET) / MAX_POS
|
||||
krw_profit = pos_size * pnl / 100
|
||||
portfolio = max(portfolio + krw_profit, MIN_BUDGET)
|
||||
total_krw += krw_profit
|
||||
|
||||
ym = buy_dt.strftime("%Y-%m")
|
||||
if ym not in monthly:
|
||||
monthly[ym] = {"trades": 0, "wins": 0, "pnl_krw": 0.0}
|
||||
monthly[ym]["trades"] += 1
|
||||
monthly[ym]["wins"] += int(is_win)
|
||||
monthly[ym]["pnl_krw"] += krw_profit
|
||||
|
||||
trade_log.append({
|
||||
"buy_dt": buy_dt, "sell_dt": sell_dt, "ticker": ticker,
|
||||
"is_win": is_win, "pnl_pct": pnl,
|
||||
"pos_size": pos_size, "krw_profit": krw_profit,
|
||||
"portfolio": portfolio,
|
||||
})
|
||||
|
||||
wins = sum(1 for t in accepted if t[0])
|
||||
return {
|
||||
"portfolio": portfolio,
|
||||
"total_krw": total_krw,
|
||||
"roi_pct": (portfolio - BUDGET) / BUDGET * 100,
|
||||
"total": len(accepted),
|
||||
"wins": wins,
|
||||
"wr": wins / len(accepted) * 100 if accepted else 0,
|
||||
"monthly": monthly,
|
||||
"trade_log": trade_log,
|
||||
}
|
||||
|
||||
|
||||
# ── 메인 ─────────────────────────────────────────────────
|
||||
def main():
|
||||
print("캐시 로드 중...")
|
||||
cache = pickle.load(open(CACHE_FILE, "rb"))
|
||||
top30 = pickle.load(open(TOP30_FILE, "rb"))
|
||||
tickers = [t for t in top30[:TOP_N] if t in cache["10m"]]
|
||||
print(f"유효 종목: {len(tickers)}개\n")
|
||||
|
||||
# 리샘플링 + 전략 실행
|
||||
all_trades = []
|
||||
wf_total_blocked = 0
|
||||
for t in tickers:
|
||||
df40 = resample_40m(cache["10m"][t])
|
||||
if len(df40) < 50:
|
||||
continue
|
||||
raw = run_vol_lead(df40, t)
|
||||
filtered, blocked = apply_wf(raw)
|
||||
wf_total_blocked += blocked
|
||||
all_trades.extend(filtered)
|
||||
|
||||
all_trades.sort(key=lambda x: x[2])
|
||||
accepted, skipped = apply_max_positions(all_trades)
|
||||
result = simulate(accepted)
|
||||
|
||||
# 최대 낙폭
|
||||
peak = BUDGET
|
||||
max_dd = 0.0
|
||||
for t in result["trade_log"]:
|
||||
peak = max(peak, t["portfolio"])
|
||||
dd = (peak - t["portfolio"]) / peak * 100
|
||||
max_dd = max(max_dd, dd)
|
||||
|
||||
# 기간 추출
|
||||
if result["trade_log"]:
|
||||
start_dt = result["trade_log"][0]["buy_dt"].strftime("%Y-%m-%d")
|
||||
end_dt = result["trade_log"][-1]["sell_dt"].strftime("%Y-%m-%d")
|
||||
else:
|
||||
start_dt = end_dt = "N/A"
|
||||
|
||||
print(f"{'='*60}")
|
||||
print(f"45일 복리 시뮬 | 40분봉 vol-lead +{THRESH}% | {len(tickers)}종목")
|
||||
print(f"기간: {start_dt} ~ {end_dt}")
|
||||
print(f"{'='*60}")
|
||||
print(f" 신호 발생: {len(all_trades) + wf_total_blocked:>4}건 (WF 차단: {wf_total_blocked}건)")
|
||||
print(f" 실제 진입: {result['total']:>4}건 ({len(skipped)}건 MAX_POS 스킵)")
|
||||
print(f" 승/패: {result['wins']}승 {result['total']-result['wins']}패"
|
||||
f" (승률 {result['wr']:.1f}%)")
|
||||
print(f" {'─'*50}")
|
||||
print(f" 초기 예산: {BUDGET:>14,}원")
|
||||
print(f" 최종 자산: {result['portfolio']:>14,.0f}원")
|
||||
print(f" 순수익: {result['total_krw']:>+14,.0f}원")
|
||||
print(f" 수익률: {result['roi_pct']:>+13.2f}%")
|
||||
print(f" 최대 낙폭: {-max_dd:>+13.2f}%"
|
||||
f" ({-max_dd / 100 * BUDGET:>+,.0f}원)")
|
||||
|
||||
monthly_krw = [m["pnl_krw"] for m in result["monthly"].values()]
|
||||
avg_monthly = sum(monthly_krw) / len(monthly_krw) if monthly_krw else 0
|
||||
print(f" 월평균 수익: {avg_monthly:>+13,.0f}원")
|
||||
|
||||
print(f"\n── 월별 수익 {'─'*40}")
|
||||
print(f" {'월':^8} │ {'거래':>4} {'승률':>5} │ {'월수익(KRW)':>14} {'누적수익(KRW)':>15}")
|
||||
cum = 0.0
|
||||
for ym, m in sorted(result["monthly"].items()):
|
||||
wr = m["wins"] / m["trades"] * 100 if m["trades"] else 0
|
||||
cum += m["pnl_krw"]
|
||||
print(f" {ym:^8} │ {m['trades']:>4}건 {wr:>4.0f}% │ "
|
||||
f"{m['pnl_krw']:>+14,.0f}원 {cum:>+14,.0f}원")
|
||||
|
||||
print(f"{'='*60}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user