199 lines
7.1 KiB
Python
199 lines
7.1 KiB
Python
"""
|
||
4/30 SHORT/SCALP 백테스트 진단 스크립트.
|
||
|
||
목적:
|
||
1) SHORT 0건의 진짜 원인 — 어느 임계치에서 거래가 발생하는지 sweep
|
||
2) SCALP -90만 trades 의 종목·시각·매매 상세
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
from database import TradeDB
|
||
import tail_engine as te
|
||
import scalping_engine as se
|
||
|
||
|
||
START = "202604300830"
|
||
END = "202604301600"
|
||
|
||
|
||
def _load_universe(db, sid: str) -> dict:
|
||
"""target_candidates_history → {slot_key: [codes]}."""
|
||
rows = db.conn.execute(
|
||
"SELECT slot_key, code FROM target_candidates_history "
|
||
"WHERE strategy_id=%s AND DATE(event_time)='2026-04-30'",
|
||
(sid,),
|
||
).fetchall()
|
||
history: dict = {}
|
||
for r in rows:
|
||
history.setdefault(r["slot_key"], []).append(r["code"])
|
||
return history
|
||
|
||
|
||
def _load_candles(db, codes, tf: int, min_n: int = 19) -> dict:
|
||
out = {}
|
||
for code in codes:
|
||
rows = db.conn.execute(
|
||
"SELECT candle_time, open, high, low, close, volume FROM ws_candles "
|
||
"WHERE timeframe=%s AND code=%s AND candle_time>=%s AND candle_time<=%s "
|
||
"AND is_confirmed=1 ORDER BY candle_time",
|
||
(tf, code, START, END),
|
||
).fetchall()
|
||
if len(rows) >= min_n:
|
||
out[code] = [dict(r) for r in rows]
|
||
return out
|
||
|
||
|
||
def _trade_pnl(t):
|
||
"""엔진별 trade dict 차이를 흡수해 pnl 추출."""
|
||
if t.get("pnl") is not None:
|
||
return float(t["pnl"])
|
||
try:
|
||
entry = float(t.get("entry") or t.get("buy_price") or 0)
|
||
exit_ = float(t.get("exit") or t.get("sell_price") or 0)
|
||
qty = int(t.get("qty") or 1)
|
||
return (exit_ - entry) * qty
|
||
except Exception:
|
||
return 0.0
|
||
|
||
|
||
def _summary(trades):
|
||
if not trades:
|
||
return "trades=0"
|
||
n = len(trades)
|
||
pnls = [_trade_pnl(t) for t in trades]
|
||
win = sum(1 for p in pnls if p > 0)
|
||
return (f"trades={n} pnl합={sum(pnls):,.0f}원 "
|
||
f"승률={win}/{n}={win/n*100:.0f}% avg={sum(pnls)/n:,.0f}원")
|
||
|
||
|
||
def diag_short(db):
|
||
print("\n" + "=" * 70)
|
||
print("SHORT (꼬리잡기) 4/30 진단")
|
||
print("=" * 70)
|
||
|
||
history = _load_universe(db, "SHORT")
|
||
codes = {c for lst in history.values() for c in lst}
|
||
print(f"universe: {len(codes)}종목, {len(history)}슬롯")
|
||
|
||
candles = _load_candles(db, codes, tf=3, min_n=19)
|
||
print(f"3분봉 적재: {len(candles)}종목 (RSI 계산 가능)")
|
||
|
||
base = te.get_tail_defaults_from_db()
|
||
base["scan_interval_min"] = 1 # SHORT slot 1분 단위
|
||
|
||
print()
|
||
print("--- 현재 DB 임계치 ---")
|
||
print(f" min_drop_rate = {base['min_drop_rate']*100:.2f}%")
|
||
print(f" min_recovery_ratio = {base['min_recovery_ratio']*100:.0f}%")
|
||
print(f" tail_ratio_min = {base['tail_ratio_min']:.2f}")
|
||
print(f" tail_pct_min = {base['tail_pct_min']*100:.2f}%")
|
||
print(f" max_daily_change = {base['max_daily_change']:.1f}%")
|
||
print(f" ma20_max_above = {base['ma20_max_above']:.1f}%")
|
||
print(f" rsi_threshold = {base['rsi_threshold']:.1f}")
|
||
print(f" max_rec_3m = {base['max_rec_3m']*100:.0f}%")
|
||
print(f" high_chase_thr = {base['high_chase_thr']:.2f}")
|
||
|
||
print()
|
||
print("--- 현재 임계치로 백테스트 ---")
|
||
trades = te.run_tail_backtest(candles, dict(base), universe_by_slot=history)
|
||
print(f" → {_summary(trades)}")
|
||
|
||
print()
|
||
print("--- 임계치 단계적 완화 sweep ---")
|
||
sweeps = [
|
||
("baseline (현재)", {}),
|
||
("MIN_DROP 1.5% / REC 25%", {"min_drop_rate": 0.015, "min_recovery_ratio": 0.25}),
|
||
("MIN_DROP 1.0% / REC 20%", {"min_drop_rate": 0.010, "min_recovery_ratio": 0.20}),
|
||
("MIN_DROP 0.5% / REC 10%", {"min_drop_rate": 0.005, "min_recovery_ratio": 0.10}),
|
||
("MIN_DROP 0% / REC 0%", {"min_drop_rate": 0.0, "min_recovery_ratio": 0.0}),
|
||
("MA20 ∞ / RSI ∞ + 위 같음", {
|
||
"min_drop_rate": 0.0, "min_recovery_ratio": 0.0,
|
||
"ma20_max_above": 999.0, "rsi_threshold": 999.0,
|
||
"max_daily_change": 999.0, "max_rec_3m": 1.0,
|
||
"high_chase_thr": 999.0,
|
||
"tail_ratio_min": 0.0, "tail_pct_min": 0.0,
|
||
}),
|
||
]
|
||
for name, ov in sweeps:
|
||
p = dict(base)
|
||
p.update(ov)
|
||
try:
|
||
t = te.run_tail_backtest(candles, p, universe_by_slot=history)
|
||
print(f" {name:35s} → {_summary(t)}")
|
||
except Exception as e:
|
||
print(f" {name:35s} → 예외: {e}")
|
||
|
||
|
||
def diag_scalp(db):
|
||
print("\n" + "=" * 70)
|
||
print("SCALP 4/30 진단")
|
||
print("=" * 70)
|
||
|
||
history = _load_universe(db, "SCALP")
|
||
codes = {c for lst in history.values() for c in lst}
|
||
print(f"universe: {len(codes)}종목, {len(history)}슬롯")
|
||
|
||
# SCALP universe slot 의 분 단위 분포 (5분 정렬 매칭률 보기)
|
||
from collections import Counter
|
||
mins = Counter(int(k[10:12]) for k in history.keys() if len(k) >= 12)
|
||
aligned5 = sum(c for m, c in mins.items() if m % 5 == 0)
|
||
total = sum(mins.values())
|
||
print(f" slot 5분 정렬 매칭률 (backtest_web 의 scan_interval_min=5 와 매칭): "
|
||
f"{aligned5}/{total} = {aligned5/total*100:.0f}%" if total else "")
|
||
|
||
candles = _load_candles(db, codes, tf=1, min_n=10)
|
||
print(f"1분봉 적재: {len(candles)}종목")
|
||
|
||
base = se.get_scalping_defaults_from_db()
|
||
|
||
# backtest_web SCALP 와 동일하게 scan_interval_min=5 로 1차 시도
|
||
base["scan_interval_min"] = 5
|
||
print()
|
||
print("--- backtest_web 과 동일 (scan_interval_min=5) ---")
|
||
trades5 = se.run_scalping_backtest(candles, dict(base), universe_by_slot=history)
|
||
print(f" → {_summary(trades5)}")
|
||
|
||
# 1분 슬롯과 1:1 매칭 (universe history 1분 키와 일치)
|
||
base["scan_interval_min"] = 1
|
||
print()
|
||
print("--- 보정: scan_interval_min=1 (universe slot 키와 일치) ---")
|
||
trades1 = se.run_scalping_backtest(candles, dict(base), universe_by_slot=history)
|
||
print(f" → {_summary(trades1)}")
|
||
|
||
trades = trades1 if trades1 else trades5
|
||
if trades:
|
||
print()
|
||
print(f"--- trades 상세 (총 {len(trades)}건) ---")
|
||
for t in trades:
|
||
t["pnl"] = _trade_pnl(t)
|
||
trades.sort(key=lambda x: x["pnl"])
|
||
print(" ▼ 손실 큰 5건:")
|
||
for t in trades[:5]:
|
||
print(f" {t.get('code')} {t.get('buy_time')}→{t.get('sell_time')} "
|
||
f"매수 {t.get('buy_price'):.0f} → 매도 {t.get('sell_price'):.0f} "
|
||
f"×{t.get('qty')} pnl={t['pnl']:,.0f} ({t.get('sell_reason')})")
|
||
print(" ▲ 수익 큰 5건:")
|
||
for t in trades[-5:][::-1]:
|
||
print(f" {t.get('code')} {t.get('buy_time')}→{t.get('sell_time')} "
|
||
f"매수 {t.get('buy_price'):.0f} → 매도 {t.get('sell_price'):.0f} "
|
||
f"×{t.get('qty')} pnl={t['pnl']:,.0f} ({t.get('sell_reason')})")
|
||
cnt = Counter(t.get("sell_reason") for t in trades)
|
||
print(f" sell_reason 분포: {dict(cnt)}")
|
||
|
||
|
||
def main():
|
||
db = TradeDB()
|
||
try:
|
||
diag_short(db)
|
||
diag_scalp(db)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|