chore: add WF/shadow/momentum analysis simulation scripts
Scripts used to analyze and validate strategy changes: - wf_cmp.py: WF window size comparison on 42 real trades - wf_cmp2.py: WF comparison extended with price_history simulation - shadow_sim.py: shadow rehabilitation sim without strategy filters - shadow_sim2.py: post-rehabilitation performance simulation - shadow_sim3.py: shadow rehabilitation sim with full strategy filters - momentum_cmp.py: momentum filter A/B comparison - trend_check.py: 2h price gain distribution analysis per ticker Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
joungmin
155
shadow_sim2.py
Normal file
155
shadow_sim2.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""Shadow 재활 이후 실제 진입 성과 시뮬레이션."""
|
||||
|
||||
import os
|
||||
from datetime import datetime
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
import oracledb
|
||||
|
||||
STOP_LOSS_PCT = float(os.getenv("STOP_LOSS_PCT", "1.5")) / 100
|
||||
TIME_STOP_HOURS = int(os.getenv("TIME_STOP_HOURS", "8"))
|
||||
TIME_STOP_MIN_PCT = float(os.getenv("TIME_STOP_MIN_GAIN_PCT", "3")) / 100
|
||||
FEE = 0.0005
|
||||
REHABILITATE_WINS = 2
|
||||
|
||||
|
||||
def get_price_series(cur, ticker, from_dt):
|
||||
cur.execute("""
|
||||
SELECT price, recorded_at FROM price_history
|
||||
WHERE ticker = :t AND recorded_at >= :dt
|
||||
ORDER BY recorded_at
|
||||
""", t=ticker, dt=from_dt)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def simulate_one(prices, buy_price, buy_dt):
|
||||
"""단일 포지션 시뮬레이션. Returns (is_win, sell_price, sell_dt, reason, pnl_pct)"""
|
||||
peak = buy_price
|
||||
for price, ts in prices:
|
||||
if price > peak:
|
||||
peak = price
|
||||
elapsed_h = (ts - buy_dt).total_seconds() / 3600
|
||||
pnl = (price - buy_price) / buy_price
|
||||
drop_from_peak = (peak - price) / peak
|
||||
if drop_from_peak >= STOP_LOSS_PCT:
|
||||
sell_pnl = (price*(1-FEE) - buy_price*(1+FEE)) / (buy_price*(1+FEE)) * 100
|
||||
return (sell_pnl > 0, price, ts, f"트레일링({pnl*100:+.1f}%)", sell_pnl)
|
||||
if elapsed_h >= TIME_STOP_HOURS and pnl < TIME_STOP_MIN_PCT:
|
||||
sell_pnl = (price*(1-FEE) - buy_price*(1+FEE)) / (buy_price*(1+FEE)) * 100
|
||||
return (sell_pnl > 0, price, ts, f"타임스탑({elapsed_h:.1f}h)", sell_pnl)
|
||||
last_price, last_ts = prices[-1]
|
||||
sell_pnl = (last_price*(1-FEE) - buy_price*(1+FEE)) / (buy_price*(1+FEE)) * 100
|
||||
return (sell_pnl > 0, last_price, last_ts, "데이터종료(보유중)", sell_pnl)
|
||||
|
||||
|
||||
def run_shadow_then_real(cur, ticker, wf_trigger_dt):
|
||||
"""shadow로 재활 후, 재활 시점 이후 실제 거래 성과 시뮬레이션."""
|
||||
prices = get_price_series(cur, ticker, wf_trigger_dt)
|
||||
if not prices:
|
||||
return None
|
||||
|
||||
# 1단계: shadow로 재활 시점 찾기
|
||||
shadow_wins = 0
|
||||
cursor_idx = 0
|
||||
rehab_dt = None
|
||||
|
||||
while cursor_idx < len(prices):
|
||||
buy_price, buy_dt = prices[cursor_idx]
|
||||
remaining = prices[cursor_idx + 1:]
|
||||
if not remaining:
|
||||
break
|
||||
|
||||
is_win, sell_price, sell_dt, reason, pnl = simulate_one(remaining, buy_price, buy_dt)
|
||||
if is_win:
|
||||
shadow_wins += 1
|
||||
else:
|
||||
shadow_wins = 0
|
||||
|
||||
if shadow_wins >= REHABILITATE_WINS:
|
||||
rehab_dt = sell_dt
|
||||
break
|
||||
|
||||
next_idx = next((i for i, (_, ts) in enumerate(prices) if ts > sell_dt), None)
|
||||
if next_idx is None:
|
||||
break
|
||||
cursor_idx = next_idx
|
||||
|
||||
if rehab_dt is None:
|
||||
return None # 재활 실패
|
||||
|
||||
# 2단계: 재활 이후 실제 거래 시뮬레이션
|
||||
print(f"\n ★ WF 해제 시점: {rehab_dt.strftime('%m-%d %H:%M')}")
|
||||
print(f" ─ 이후 실제 진입 시뮬레이션 ─")
|
||||
|
||||
post_prices = get_price_series(cur, ticker, rehab_dt)
|
||||
if not post_prices:
|
||||
print(" → 재활 이후 가격 데이터 없음")
|
||||
return
|
||||
|
||||
cursor_idx = 0
|
||||
trade_no = 0
|
||||
wins = 0
|
||||
total_pnl = 0.0
|
||||
|
||||
while cursor_idx < len(post_prices):
|
||||
buy_price, buy_dt = post_prices[cursor_idx]
|
||||
remaining = post_prices[cursor_idx + 1:]
|
||||
if not remaining:
|
||||
break
|
||||
|
||||
is_win, sell_price, sell_dt, reason, pnl = simulate_one(remaining, buy_price, buy_dt)
|
||||
trade_no += 1
|
||||
if is_win:
|
||||
wins += 1
|
||||
total_pnl += pnl
|
||||
|
||||
mark = "✅" if is_win else "❌"
|
||||
print(f" 실제#{trade_no}: {buy_price:.4f}→{sell_price:.4f}원 | {mark} {pnl:+.2f}% | {reason} ({sell_dt.strftime('%m-%d %H:%M')})")
|
||||
|
||||
next_idx = next((i for i, (_, ts) in enumerate(post_prices) if ts > sell_dt), None)
|
||||
if next_idx is None:
|
||||
break
|
||||
cursor_idx = next_idx
|
||||
|
||||
if trade_no > 0:
|
||||
wr = wins / trade_no * 100
|
||||
print(f"\n 📊 재활 후 성과: {trade_no}건 중 {wins}승 | 승률={wr:.0f}% | 누적PnL={total_pnl:+.2f}%")
|
||||
return {'trades': trade_no, 'wins': wins, 'win_rate': wr, 'total_pnl': total_pnl}
|
||||
return None
|
||||
|
||||
|
||||
def run():
|
||||
conn = oracledb.connect(
|
||||
user=os.getenv('ORACLE_USER'),
|
||||
password=os.getenv('ORACLE_PASSWORD'),
|
||||
dsn=os.getenv('ORACLE_DSN'),
|
||||
config_dir=os.getenv('ORACLE_WALLET')
|
||||
)
|
||||
cur = conn.cursor()
|
||||
|
||||
results = {}
|
||||
for ticker in ['KRW-DKA', 'KRW-LAYER', 'KRW-SIGN']:
|
||||
cur.execute("""
|
||||
SELECT traded_at FROM trade_results
|
||||
WHERE ticker = :t ORDER BY traded_at
|
||||
""", t=ticker)
|
||||
rows = cur.fetchall()
|
||||
wf_trigger_dt = rows[4][0]
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"[{ticker}] WF차단 발동: {wf_trigger_dt.strftime('%m-%d %H:%M')}")
|
||||
r = run_shadow_then_real(cur, ticker, wf_trigger_dt)
|
||||
if r:
|
||||
results[ticker] = r
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print("전체 요약")
|
||||
print(f"{'='*60}")
|
||||
for ticker, r in results.items():
|
||||
print(f"{ticker}: 재활 후 {r['trades']}건 | 승률={r['win_rate']:.0f}% | 누적={r['total_pnl']:+.2f}%")
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
Reference in New Issue
Block a user