6e0c4508fa
- tick_trader.py를 Controller로 축소, 로직을 3개 모듈로 분리: - core/signal.py: 시그널 감지, 지표 계산 (calc_vr, calc_atr, detect_signal) - core/order.py: Upbit 주문 실행 (매수/매도/취소/조회) - core/position_manager.py: 포지션 관리, DB sync, 복구, 청산 조건 - type hints, Google docstring, 구체적 예외 타입 적용 - 50줄 초과 함수 분리 (process_signal, restore_positions) - 미사용 파일 58개 archive/ 폴더로 이동 - README.md 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
475 lines
18 KiB
Python
475 lines
18 KiB
Python
"""레짐 기반 1년 시뮬레이션 — BULL 진입 vs Bear차단 vs 필터없음.
|
|
|
|
sim_45m40.py 검증된 코어 로직 기반.
|
|
데이터: data/sim1y_cache.pkl (10분봉 1년치)
|
|
data/fng_1y.json (F&G 1년치)
|
|
|
|
비교 구성:
|
|
1. 필터 없음 — 레짐/F&G 무관 진입
|
|
2. BEAR 차단 — 레짐 score < -0.5% 이면 차단 (현재 전략)
|
|
3. BULL 진입만 — 레짐 score ≥ 1.5% 일 때만 진입 ← 사용자 제안
|
|
4. BULL 진입 + F&G≥41 — BULL 조건에 F&G 필터 추가
|
|
"""
|
|
import os as _os, sys as _sys
|
|
_sys.path.insert(0, _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))))
|
|
|
|
import json
|
|
import pickle
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(dotenv_path=Path(__file__).parent.parent / ".env")
|
|
|
|
# ── 데이터 경로 ───────────────────────────────────────────
|
|
CACHE_FILE = Path(__file__).parent.parent / "data" / "sim1y_cache.pkl"
|
|
FNG_FILE = Path(__file__).parent.parent / "data" / "fng_1y.json"
|
|
TOP_N = 20
|
|
|
|
# ── 전략 파라미터 (sim_45m40.py 동일) ────────────────────
|
|
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_DEFAULT = 2.0 # 기본 (NEUTRAL / 필터없음)
|
|
VOL_MULT_BULL = 1.5 # BULL 레짐 완화
|
|
QUIET_PCT = 2.0
|
|
THRESH = 4.8 # sim_45m40.py 기준값
|
|
|
|
# 40분봉 봉수 환산
|
|
LOCAL_VOL_N = 7 # 5h
|
|
QUIET_N = 3 # 2h
|
|
SIGNAL_TO_N = 12 # 8h
|
|
ATR_N = 7
|
|
TS_N = 12 # 8h (타임스탑)
|
|
REGIME_N = 3 # 2h (레짐 추세)
|
|
|
|
# 레짐 임계
|
|
BEAR_THRESHOLD = -0.5
|
|
BULL_THRESHOLD = 1.5
|
|
|
|
# 레짐 계산 가중치
|
|
REGIME_WEIGHTS = {"KRW-BTC": 0.40, "KRW-ETH": 0.30,
|
|
"KRW-SOL": 0.15, "KRW-XRP": 0.15}
|
|
|
|
# WF 파라미터
|
|
WF_WINDOW = 4
|
|
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"])
|
|
)
|
|
|
|
|
|
# ── 레짐 시리즈 ──────────────────────────────────────────
|
|
def build_regime_series(dfs40: dict) -> pd.Series:
|
|
weighted = None
|
|
for ticker, w in REGIME_WEIGHTS.items():
|
|
if ticker not in dfs40:
|
|
continue
|
|
pct = dfs40[ticker]["close"].pct_change(REGIME_N) * 100
|
|
weighted = pct * w if weighted is None else weighted.add(pct * w, fill_value=0.0)
|
|
return weighted if weighted is not None else pd.Series(dtype=float)
|
|
|
|
|
|
# ── 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
|
|
|
|
|
|
# ── 전략 실행 ─────────────────────────────────────────────
|
|
def run_strategy(df: pd.DataFrame, ticker: str,
|
|
regime_series: pd.Series, fng_map: dict,
|
|
mode: str) -> list:
|
|
"""
|
|
mode:
|
|
'none' — 레짐/F&G 필터 없음
|
|
'bear_off' — BEAR 차단만 (score < BEAR_THRESHOLD 시 스킵)
|
|
'bull_only'— BULL 진입만 (score >= BULL_THRESHOLD 일 때만)
|
|
'bull_fng' — BULL + F&G≥41
|
|
"""
|
|
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):
|
|
ts = df.index[i]
|
|
row = df.iloc[i]
|
|
cur = row["close"]
|
|
|
|
# ── 포지션 보유 중 ────────────────────────────────
|
|
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
|
|
|
|
# ── 레짐 스코어 계산 ─────────────────────────────
|
|
score = 0.0
|
|
if not regime_series.empty and ts in regime_series.index:
|
|
v = regime_series.loc[ts]
|
|
score = float(v) if not pd.isna(v) else 0.0
|
|
|
|
# ── 모드별 진입 필터 ─────────────────────────────
|
|
if mode == "bear_off":
|
|
if score < BEAR_THRESHOLD:
|
|
sig_i = sig_p = None
|
|
i += 1
|
|
continue
|
|
vol_mult = VOL_MULT_BULL if score >= BULL_THRESHOLD else VOL_MULT_DEFAULT
|
|
|
|
elif mode == "bull_only":
|
|
if score < BULL_THRESHOLD:
|
|
sig_i = sig_p = None
|
|
i += 1
|
|
continue
|
|
vol_mult = VOL_MULT_BULL
|
|
|
|
elif mode == "bull_fng":
|
|
if score < BULL_THRESHOLD:
|
|
sig_i = sig_p = None
|
|
i += 1
|
|
continue
|
|
date_str = ts.strftime("%Y-%m-%d")
|
|
fv = fng_map.get(date_str, 50) if fng_map else 50
|
|
if fv < 41:
|
|
sig_i = sig_p = None
|
|
i += 1
|
|
continue
|
|
vol_mult = VOL_MULT_BULL
|
|
|
|
else: # 'none'
|
|
vol_mult = VOL_MULT_DEFAULT
|
|
|
|
# ── 신호 타임아웃 ─────────────────────────────────
|
|
if sig_i is not None and (i - sig_i) > SIGNAL_TO_N:
|
|
sig_i = sig_p = None
|
|
|
|
# ── 신호 있음: 진입 체크 ──────────────────────────
|
|
if sig_i is not None:
|
|
move_pct = (cur - sig_p) / sig_p * 100
|
|
if cur < sig_p:
|
|
sig_i = sig_p = None
|
|
elif move_pct >= THRESH:
|
|
in_pos = True
|
|
buy_idx = i
|
|
buy_price = cur
|
|
stop_pct = calc_atr(df, i)
|
|
sig_i = sig_p = None
|
|
i += 1
|
|
continue
|
|
|
|
# ── 신호 없음: 축적 조건 체크 ────────────────────
|
|
vol_p = df.iloc[i - 1]["volume"]
|
|
vol_avg = df.iloc[i - 1 - LOCAL_VOL_N: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(cur - close_qh) / close_qh * 100 if close_qh > 0 else 999
|
|
quiet = chg_qh < QUIET_PCT
|
|
spike = vol_r >= vol_mult
|
|
|
|
if quiet and spike:
|
|
if sig_i is None:
|
|
sig_i = i
|
|
sig_p = cur
|
|
else:
|
|
if sig_i is not None and cur < sig_p:
|
|
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 run_compound(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, "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 print_result(label: str, result: dict, skipped: int, wf_blocked: int):
|
|
r = result
|
|
peak = BUDGET
|
|
max_dd = 0.0
|
|
for t in r["trade_log"]:
|
|
peak = max(peak, t["portfolio"])
|
|
dd = (peak - t["portfolio"]) / peak * 100
|
|
max_dd = max(max_dd, dd)
|
|
|
|
monthly_krw = [m["pnl_krw"] for m in r["monthly"].values()]
|
|
avg_m = sum(monthly_krw) / len(monthly_krw) if monthly_krw else 0
|
|
|
|
total = r["total"]
|
|
wins = r["wins"]
|
|
|
|
print(f"\n{'─'*60}")
|
|
print(f" [{label}]")
|
|
print(f" 진입: {total}건 (WF차단: {wf_blocked} / MAX_POS스킵: {skipped})")
|
|
if total:
|
|
print(f" 승패: {wins}승 {total-wins}패 (승률 {r['wr']:.1f}%)")
|
|
print(f" 초기 예산: {BUDGET:>15,}원")
|
|
print(f" 최종 자산: {r['portfolio']:>15,.0f}원")
|
|
print(f" 순수익: {r['total_krw']:>+15,.0f}원")
|
|
print(f" 수익률: {r['roi_pct']:>+14.2f}%")
|
|
print(f" 최대 낙폭: {-max_dd:>+14.2f}%")
|
|
print(f" 월평균 수익: {avg_m:>+13,.0f}원")
|
|
|
|
|
|
def print_monthly(result: dict):
|
|
print(f" {'월':^8} │ {'거래':>4} {'승률':>5} │ {'월수익(KRW)':>13} {'누적수익(KRW)':>14}")
|
|
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']:>+13,.0f}원 {cum:>+14,.0f}원")
|
|
|
|
|
|
# ── 메인 ─────────────────────────────────────────────────
|
|
def main():
|
|
# ── 데이터 로드 ───────────────────────────────────────
|
|
if not CACHE_FILE.exists():
|
|
print(f"[오류] 캐시 없음: {CACHE_FILE}")
|
|
print(" 먼저 tests/collect_1y_data.py 를 실행하세요.")
|
|
return
|
|
|
|
print("캐시 로드 중...")
|
|
cache = pickle.load(open(CACHE_FILE, "rb"))
|
|
all_tickers = list(cache["10m"].keys())[:TOP_N]
|
|
tickers = [t for t in all_tickers if len(cache["10m"][t]) > 500]
|
|
print(f" 유효 종목: {len(tickers)}개")
|
|
|
|
# F&G 로드
|
|
fng_map: dict = {}
|
|
if FNG_FILE.exists():
|
|
fng_map = json.load(open(FNG_FILE))
|
|
fng_dates = sorted(fng_map.keys())
|
|
print(f" F&G: {fng_dates[0]} ~ {fng_dates[-1]} ({len(fng_map)}일)")
|
|
else:
|
|
print(" [경고] F&G 데이터 없음")
|
|
|
|
# 리샘플링
|
|
dfs40 = {t: resample_40m(cache["10m"][t]) for t in tickers}
|
|
|
|
# 레짐 시리즈
|
|
regime_series = build_regime_series(dfs40)
|
|
|
|
# 기간 정보
|
|
sample = next(iter(dfs40.values()))
|
|
start_dt = sample.index[0].strftime("%Y-%m-%d")
|
|
end_dt = sample.index[-1].strftime("%Y-%m-%d")
|
|
|
|
# 레짐 분포 계산
|
|
if not regime_series.empty:
|
|
valid = regime_series.dropna()
|
|
bull_cnt = (valid >= BULL_THRESHOLD).sum()
|
|
bear_cnt = (valid < BEAR_THRESHOLD).sum()
|
|
neut_cnt = len(valid) - bull_cnt - bear_cnt
|
|
total_cnt = len(valid)
|
|
print(f"\n 레짐 분포 ({total_cnt}봉 기준):")
|
|
print(f" BULL (≥{BULL_THRESHOLD}%) : {bull_cnt:>6}봉 ({bull_cnt/total_cnt*100:.1f}%)")
|
|
print(f" NEUTRAL : {neut_cnt:>6}봉 ({neut_cnt/total_cnt*100:.1f}%)")
|
|
print(f" BEAR (<{BEAR_THRESHOLD}%) : {bear_cnt:>6}봉 ({bear_cnt/total_cnt*100:.1f}%)")
|
|
|
|
# F&G 분포 (해당 기간)
|
|
if fng_map:
|
|
period_fng = {k: v for k, v in fng_map.items() if start_dt <= k <= end_dt}
|
|
zones = {"극공포(≤25)": 0, "공포(26~40)": 0, "중립+(≥41)": 0}
|
|
for v in period_fng.values():
|
|
if v <= 25: zones["극공포(≤25)"] += 1
|
|
elif v <= 40: zones["공포(26~40)"] += 1
|
|
else: zones["중립+(≥41)"] += 1
|
|
tot = sum(zones.values())
|
|
print(f"\n F&G 분포 (동 기간 {tot}일):")
|
|
for name, cnt in zones.items():
|
|
print(f" {name:12} {cnt:>3}일 ({cnt/tot*100:.1f}%)")
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" 레짐 BULL 진입 시뮬 | 1년 | {len(tickers)}종목 | 40분봉")
|
|
print(f" 기간: {start_dt} ~ {end_dt}")
|
|
print(f"{'='*60}")
|
|
|
|
# ── 4가지 시뮬 실행 ───────────────────────────────────
|
|
CONFIGS = [
|
|
("none", "필터 없음"),
|
|
("bear_off", "BEAR 차단 (현재)"),
|
|
("bull_only","BULL 진입만"),
|
|
("bull_fng", "BULL + F&G≥41"),
|
|
]
|
|
|
|
results = {}
|
|
for mode, label in CONFIGS:
|
|
all_trades = []
|
|
wf_total = 0
|
|
for ticker in tickers:
|
|
df40 = dfs40[ticker]
|
|
raw = run_strategy(df40, ticker, regime_series, fng_map, mode)
|
|
filtered, blocked = apply_wf(raw)
|
|
wf_total += blocked
|
|
all_trades.extend(filtered)
|
|
|
|
all_trades.sort(key=lambda x: x[2])
|
|
accepted, skipped = apply_max_positions(all_trades)
|
|
result = run_compound(accepted)
|
|
results[label] = result
|
|
print_result(label, result, len(skipped), wf_total)
|
|
|
|
# ── 요약 비교 ─────────────────────────────────────────
|
|
print(f"\n{'='*60}")
|
|
print(f" 요약 비교")
|
|
print(f"{'='*60}")
|
|
print(f" {'구성':<22} {'진입':>5} {'승률':>6} {'수익률':>8} {'순수익(KRW)':>14} {'낙폭':>7}")
|
|
print(f" {'─'*58}")
|
|
for mode, label in CONFIGS:
|
|
r = results[label]
|
|
total = r["total"]
|
|
if total == 0:
|
|
print(f" {label:<22} {'진입없음':>34}")
|
|
continue
|
|
peak = BUDGET
|
|
max_dd = 0.0
|
|
for t in r["trade_log"]:
|
|
peak = max(peak, t["portfolio"])
|
|
dd = (peak - t["portfolio"]) / peak * 100
|
|
max_dd = max(max_dd, dd)
|
|
print(
|
|
f" {label:<22} {total:>5}건 {r['wr']:>5.1f}% "
|
|
f"{r['roi_pct']:>+7.2f}% {r['total_krw']:>+13,.0f}원 -{max_dd:.1f}%"
|
|
)
|
|
|
|
# ── 월별 상세 (BULL 진입만) ───────────────────────────
|
|
print(f"\n{'='*60}")
|
|
print(f" 월별 상세 — BULL 진입만")
|
|
print(f"{'='*60}")
|
|
print_monthly(results["BULL 진입만"])
|
|
print(f"\n 월별 상세 — BEAR 차단 (현재)")
|
|
print_monthly(results["BEAR 차단 (현재)"])
|
|
print(f"{'='*60}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|