"""캐스케이드 limit 주문 전략 시뮬 (30일). 전략: ① bars[0:2] → 2봉, +2% limit (trail 없음) ② bars[2:5] → 3봉, +1% limit (trail 없음) ③ bars[5:5+last_n] → last_n봉, +0.5% limit (trail 없음) ④ bars[5+last_n:] → 기존전략 (TP2% + ATR Trail Stop) """ import sys, os from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '.env')) import oracledb LOOKBACK_DAYS = 30 VOL_LOOKBACK = 61 ATR_LOOKBACK = 28 VOL_MIN = 8.0 ATR_MULT = 1.0 ATR_MIN_R = 0.030 ATR_MAX_R = 0.050 MAX_TRAIL_BARS = 240 BUDGET = 15_000_000 MAX_POS = 3 PER_POS = BUDGET // MAX_POS FEE = 0.0005 TICKERS = [ 'KRW-XRP','KRW-BTC','KRW-ETH','KRW-SOL','KRW-DOGE', 'KRW-ADA','KRW-SUI','KRW-NEAR','KRW-KAVA','KRW-SXP', 'KRW-AKT','KRW-SONIC','KRW-IP','KRW-ORBS','KRW-VIRTUAL', 'KRW-BARD','KRW-XPL','KRW-KITE','KRW-ENSO','KRW-0G', ] _TK = ",".join(f"'{t}'" for t in TICKERS) def get_conn(): kwargs = dict(user=os.environ["ORACLE_USER"], password=os.environ["ORACLE_PASSWORD"], dsn=os.environ["ORACLE_DSN"]) if w := os.environ.get("ORACLE_WALLET"): kwargs["config_dir"] = w return oracledb.connect(**kwargs) SIGNAL_SQL = f""" WITH base AS ( SELECT ticker, ts, open_p, close_p, high_p, low_p, volume_p, LAG(close_p,1) OVER (PARTITION BY ticker ORDER BY ts) pc1, LAG(open_p,1) OVER (PARTITION BY ticker ORDER BY ts) po1, LAG(close_p,2) OVER (PARTITION BY ticker ORDER BY ts) pc2, LAG(open_p,2) OVER (PARTITION BY ticker ORDER BY ts) po2, GREATEST(high_p-low_p, ABS(high_p-LAG(close_p,1) OVER (PARTITION BY ticker ORDER BY ts)), ABS(low_p -LAG(close_p,1) OVER (PARTITION BY ticker ORDER BY ts))) tr FROM backtest_ohlcv WHERE interval_cd='minute1' AND ts >= TO_TIMESTAMP(:ws,'YYYY-MM-DD HH24:MI:SS') AND ticker IN ({_TK}) ), ind AS ( SELECT ticker, ts, open_p, close_p, high_p, low_p, volume_p / NULLIF(AVG(volume_p) OVER ( PARTITION BY ticker ORDER BY ts ROWS BETWEEN {VOL_LOOKBACK} PRECEDING AND 2 PRECEDING),0) vr0, LAG(volume_p,1) OVER (PARTITION BY ticker ORDER BY ts) / NULLIF(AVG(volume_p) OVER ( PARTITION BY ticker ORDER BY ts ROWS BETWEEN {VOL_LOOKBACK} PRECEDING AND 2 PRECEDING),0) vr1, LAG(volume_p,2) OVER (PARTITION BY ticker ORDER BY ts) / NULLIF(AVG(volume_p) OVER ( PARTITION BY ticker ORDER BY ts ROWS BETWEEN {VOL_LOOKBACK} PRECEDING AND 2 PRECEDING),0) vr2, pc1,po1,pc2,po2, AVG(tr) OVER (PARTITION BY ticker ORDER BY ts ROWS BETWEEN {ATR_LOOKBACK} PRECEDING AND 1 PRECEDING) / NULLIF(pc1,0) atr_raw FROM base ) SELECT ticker, ts, vr0, vr1, vr2, atr_raw FROM ind WHERE ts >= TO_TIMESTAMP(:cs,'YYYY-MM-DD HH24:MI:SS') AND vr0 >= {VOL_MIN} AND close_p>open_p AND pc1>po1 AND pc2>po2 AND close_p>pc1 AND pc1>pc2 AND vr0>vr1 AND vr1>vr2 ORDER BY ticker, ts """ def fetch_signals(cur, warmup_since, check_since): cur.execute(SIGNAL_SQL, {'ws': warmup_since, 'cs': check_since}) rows = cur.fetchall() signals = [] for row in rows: ticker, sig_ts, vr0, vr1, vr2, atr_raw = row cur.execute( """SELECT close_p, ts FROM backtest_ohlcv WHERE ticker=:t AND interval_cd='minute1' AND ts > :sig AND ts <= :sig + INTERVAL '3' MINUTE ORDER BY ts FETCH FIRST 1 ROWS ONLY""", {'t': ticker, 'sig': sig_ts} ) er = cur.fetchone() if not er: continue ep, ets = float(er[0]), er[1] cur.execute( """SELECT ts, close_p, high_p, low_p FROM backtest_ohlcv WHERE ticker=:t AND interval_cd='minute1' AND ts >= :entry ORDER BY ts FETCH FIRST :n ROWS ONLY""", {'t': ticker, 'entry': ets, 'n': MAX_TRAIL_BARS + 1} ) bars = [(r[0], float(r[1]), float(r[2]), float(r[3])) for r in cur.fetchall()] if not bars: continue signals.append({ 'ticker': ticker, 'entry_ts': ets, 'entry_price': ep, 'atr_raw': float(atr_raw) if atr_raw else 0.0, 'bars': bars, }) signals.sort(key=lambda x: x['entry_ts']) return signals def sim_tp_trail(bars, ep, ar, tp_r=0.02): """기본 전략: TP + Trail Stop.""" stop = max(ATR_MIN_R, min(ATR_MAX_R, ar * ATR_MULT)) if ar > 0 else ATR_MAX_R tp = ep * (1 + tp_r) peak = ep for i, (ts, cp, hp, lp) in enumerate(bars): if hp >= tp: return dict(status='TP2%', exit_ts=ts, exit_price=tp, pnl=tp_r * 100, held=i + 1) peak = max(peak, cp) if (peak - cp) / peak >= stop: return dict(status='트레일손절', exit_ts=ts, exit_price=cp, pnl=(cp - ep) / ep * 100, held=i + 1) lts, lcp = bars[-1][0], bars[-1][1] return dict(status='타임아웃' if len(bars) >= MAX_TRAIL_BARS else '진행중', exit_ts=lts, exit_price=lcp, pnl=(lcp - ep) / ep * 100, held=len(bars)) def sim_cascade(bars, ep, ar, last_n): """ ① bars[0:2] → 2봉, +2% limit ② bars[2:5] → 3봉, +1% limit ③ bars[5:5+last_n] → last_n봉, +0.5% limit ④ bars[5+last_n:] → 기존전략 (TP2% + Trail Stop) """ stages = [ (0, 2, 0.020, f'①2봉2%'), (2, 5, 0.010, f'②3봉1%'), (5, 5 + last_n, 0.005, f'③{last_n}봉0.5%'), ] for start, end, lr, tag in stages: lp = ep * (1 + lr) for i, (ts, cp, hp, _) in enumerate(bars[start:end]): if hp >= lp: return dict(status=tag, exit_ts=ts, exit_price=lp, pnl=lr * 100, held=start + i + 1) offset = 5 + last_n fb = sim_tp_trail(bars[offset:] or bars[-1:], ep, ar) fb['held'] += offset fb['status'] = '④기존→' + fb['status'] return fb def sim_limit_then_trail(bars, ep, ar, n_bars=2, limit_r=0.005, tp_r=0.02): """단순 limit: N봉 내 체결 안되면 TP/Trail.""" lp = ep * (1 + limit_r) for i, (ts, cp, hp, _) in enumerate(bars[:n_bars]): if hp >= lp: return dict(status=f'limit{limit_r*100:.1f}%', exit_ts=ts, exit_price=lp, pnl=limit_r * 100, held=i + 1) fb = sim_tp_trail(bars[n_bars:] or bars[-1:], ep, ar, tp_r) fb['held'] += n_bars fb['status'] = '미체결→' + fb['status'] return fb def pos_limit(sim): opens, taken, skipped = [], [], [] for r in sim: opens = [ex for ex in opens if ex > r['entry_ts']] if len(opens) < MAX_POS: opens.append(r['exit_ts']) taken.append(r) else: skipped.append(r) return taken, skipped def krw(r): return PER_POS * (r['pnl'] / 100) - PER_POS * FEE * 2 def print_cascade_detail(taken, last_n, label): stage_tags = ['①2봉2%', '②3봉1%', f'③{last_n}봉0.5%'] stage_lr = [0.020, 0.010, 0.005] print(f"\n{'━'*70}") print(f" {label}") print(f" 총 {len(taken)}건 승률 {sum(1 for r in taken if r['pnl']>0)/len(taken)*100:.0f}% " f"합산 {sum(krw(r) for r in taken):+,.0f}원") print(f"{'━'*70}") for tag, lr in zip(stage_tags, stage_lr): grp = [r for r in taken if r['status'] == tag] if not grp: continue total = sum(krw(r) for r in grp) avg = total / len(grp) print(f" ┌─ {tag}: {len(grp):3d}건 avg {avg:+,.0f}원/건 소계 {total:+,.0f}원") # ④ 기존전략 하위 분류 fb_grp = [r for r in taken if r['status'].startswith('④기존→')] if fb_grp: print(f" └─ ④기존전략 (미체결 후): {len(fb_grp)}건") for sub in ['TP2%', '트레일손절', '타임아웃', '진행중']: sub_grp = [r for r in fb_grp if r['status'].endswith(sub)] if not sub_grp: continue total = sum(krw(r) for r in sub_grp) avg = total / len(sub_grp) print(f" {'▲' if total>0 else '▼'} {sub:8s}: {len(sub_grp):3d}건 " f"avg {avg:+,.0f}원/건 소계 {total:+,.0f}원") print() def main(): now = datetime.now() check_since = (now - timedelta(days=LOOKBACK_DAYS)).strftime('%Y-%m-%d 00:00:00') warmup_since = (now - timedelta(days=LOOKBACK_DAYS + 1)).strftime('%Y-%m-%d 00:00:00') conn = get_conn() cur = conn.cursor() cur.arraysize = 10000 print(f"=== 캐스케이드 limit 전략 시뮬 ===") print(f"기간: {check_since[:10]} ~ {now.strftime('%Y-%m-%d')} (30일)\n") signals = fetch_signals(cur, warmup_since, check_since) print(f"시그널 {len(signals)}건\n") # ── 기준선: 현재전략 ───────────────────────────────────────────────────── base_sim = [] for s in signals: r = sim_tp_trail(s['bars'], s['entry_price'], s['atr_raw']) base_sim.append({**s, **r}) base_taken, _ = pos_limit(base_sim) base_total = sum(krw(r) for r in base_taken) base_wr = sum(1 for r in base_taken if r['pnl'] > 0) / len(base_taken) * 100 print(f"{'━'*70}") print(f" [기준] 현재전략 TP2%+Trail: {len(base_taken)}건 " f"승률 {base_wr:.0f}% 합산 {base_total:+,.0f}원") # ── 비교: limit 0.5%/2봉 → TP/Trail ───────────────────────────────────── lim_sim = [] for s in signals: r = sim_limit_then_trail(s['bars'], s['entry_price'], s['atr_raw']) lim_sim.append({**s, **r}) lim_taken, _ = pos_limit(lim_sim) lim_total = sum(krw(r) for r in lim_taken) lim_wr = sum(1 for r in lim_taken if r['pnl'] > 0) / len(lim_taken) * 100 print(f" [비교] limit 0.5%/2봉→TP/Trail: {len(lim_taken)}건 " f"승률 {lim_wr:.0f}% 합산 {lim_total:+,.0f}원") print(f"{'━'*70}\n") # ── 캐스케이드 (15봉 / 30봉) ───────────────────────────────────────────── for last_n in [15, 30]: label = f"cascade ①2봉+2% → ②3봉+1% → ③{last_n}봉+0.5% → ④기존전략" csim = [] for s in signals: r = sim_cascade(s['bars'], s['entry_price'], s['atr_raw'], last_n) csim.append({**s, **r}) taken, _ = pos_limit(csim) print_cascade_detail(taken, last_n, label) conn.close() if __name__ == '__main__': main()