165 lines
7.1 KiB
Python
165 lines
7.1 KiB
Python
"""
|
|
어깨매도(Shoulder Cut) 백테스트 및 원인 분석
|
|
|
|
[문제] 어깨매도가 마이너스 수익률로 체결되는 이유
|
|
============================================================
|
|
현재 로직 (kis_short_ver2.py):
|
|
- 어깨매도 = "보유 중 고점(max_price) 대비 3%(SHOULDER_CUT_PCT) 이상 하락 시"
|
|
수익/손실 불문하고 즉시 매도.
|
|
- max_price 초기값 = 매수가(buy_price). 루프마다 현재가로 갱신.
|
|
|
|
왜 마이너스로 나가는가?
|
|
1) 매수 후 한 번도 안 올랐을 때
|
|
→ max_price = buy_price 유지 → 고점 대비 3% 하락 = 매수가 대비 -3% 하락
|
|
→ 이미 손실 구간에서 어깨매도 조건만 만족해 매도됨.
|
|
|
|
2) 조금 올랐다가 3% 빠졌을 때
|
|
→ 예: 매수가 10,000원 → 10,100원 갱신(고점) → 9,800원으로 하락(고점 대비 ~3%)
|
|
→ 어깨매도 체결 시 매수가 대비 -2% (손실).
|
|
|
|
결론: "그 전에 팔아야 한다" = 수익이 났을 때 먼저 팔아야 하는데,
|
|
현재는 "고점 대비 N% 하락"만 보고 있어서, 고점이 매수가 근처면
|
|
손실로 어깨매도되는 것이 현재 설계 그대로 동작한 결과임.
|
|
|
|
개선 방향 (백테스트에서 검증 대상):
|
|
- (A) 어깨매도를 "수익일 때만" 적용: profit_pct > 0 일 때만 고점 대비 N% 하락 시 매도.
|
|
- (B) 고점이 매수가 + X% 이상일 때만 어깨매도 적용 (일단 수익 구간 진입한 경우만).
|
|
"""
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from collections import defaultdict
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
DB_PATH = SCRIPT_DIR / "quant_bot.db"
|
|
|
|
|
|
def get_conn():
|
|
return sqlite3.connect(DB_PATH)
|
|
|
|
|
|
def report_shoulder_sell_analysis():
|
|
"""DB trade_history에서 어깨매도 건만 추출해 통계 및 원인 보고."""
|
|
conn = get_conn()
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
# 어깨매도만 필터
|
|
rows = conn.execute("""
|
|
SELECT code, name, buy_price, sell_price, qty, profit_rate, realized_pnl,
|
|
buy_date, sell_date, sell_reason
|
|
FROM trade_history
|
|
WHERE sell_reason = '어깨매도'
|
|
ORDER BY sell_date DESC
|
|
""").fetchall()
|
|
|
|
conn.close()
|
|
|
|
if not rows:
|
|
print("[어깨매도] trade_history에 '어깨매도' 건이 없습니다.")
|
|
return []
|
|
|
|
total = len(rows)
|
|
wins = [r for r in rows if r["profit_rate"] > 0]
|
|
losses = [r for r in rows if r["profit_rate"] <= 0]
|
|
win_count = len(wins)
|
|
loss_count = len(losses)
|
|
avg_profit = sum(r["profit_rate"] for r in rows) / total
|
|
avg_loss_when_loss = sum(r["profit_rate"] for r in losses) / loss_count if losses else 0
|
|
total_pnl = sum(r["realized_pnl"] for r in rows)
|
|
|
|
print("=" * 60)
|
|
print("📊 어깨매도(Shoulder Cut) 실거래 통계 (trade_history)")
|
|
print("=" * 60)
|
|
print(f" 총 건수: {total}")
|
|
print(f" 익절 건수: {win_count} ({100*win_count/total:.1f}%)")
|
|
print(f" 손실 건수: {loss_count} ({100*loss_count/total:.1f}%)")
|
|
print(f" 평균 수익률: {avg_profit:.2f}%")
|
|
if losses:
|
|
print(f" 손실 건 평균 수익률: {avg_loss_when_loss:.2f}%")
|
|
print(f" 실현 손익 합계: {total_pnl:+,.0f}원")
|
|
print()
|
|
|
|
print("📌 왜 어깨매도가 마이너스로 나가는가?")
|
|
print(" - 조건: 보유 고점(max_price) 대비 3% 하락 시 무조건 매도.")
|
|
print(" - max_price 초기값 = 매수가 → 올라본 적 없으면 '매수가=고점'에서 3% 하락 = 이미 -3% 근처.")
|
|
print(" - 조금 올랐다가 3% 빠져도, 매수가 대비하면 손실일 수 있음.")
|
|
print(" → 개선: 수익일 때만 어깨매도 적용(profit_pct>0) 또는 고점이 매수가+X% 이상일 때만 적용.")
|
|
print()
|
|
|
|
return rows
|
|
|
|
|
|
def backtest_shoulder_rule_current(rows):
|
|
"""현재 규칙 그대로: 이미 DB에 기록된 결과 요약 (재현 확인)."""
|
|
if not rows:
|
|
return
|
|
print("=" * 60)
|
|
print("🔬 백테스트 1: 현재 규칙 (고점 대비 3% 하락 시 무조건 매도)")
|
|
print("=" * 60)
|
|
print(" → 실거래와 동일. 위 실거래 통계가 곧 현재 규칙 결과.")
|
|
print()
|
|
|
|
|
|
def backtest_shoulder_rule_profit_only(rows):
|
|
"""
|
|
개선안 A: 수익일 때만 어깨매도 적용.
|
|
(실제로는 매도 시점의 profit_rate만 있으므로, '그때 수익이었으면 어깨매도 적용했을 것'으로 가정.)
|
|
여기서는 "손실로 어깨매도된 건"을 제외하면 몇 건이었을지, 그때 손익이 얼마였을지 보여줌.
|
|
"""
|
|
if not rows:
|
|
return
|
|
# 손실로 어깨매도된 건 = 개선안 적용 시 어깨매도로 나가지 않았을 건
|
|
would_skip = [r for r in rows if r["profit_rate"] <= 0]
|
|
would_apply = [r for r in rows if r["profit_rate"] > 0]
|
|
|
|
print("=" * 60)
|
|
print("🔬 백테스트 2: 개선안 A — 수익일 때만 어깨매도 적용")
|
|
print("=" * 60)
|
|
print(f" 현재 규칙으로 손실로 어깨매도된 건: {len(would_skip)}건")
|
|
print(f" → 개선안 적용 시 이 건들은 어깨매도 조건에서 제외됨 (다른 손절/익절로 나갔을 가능성).")
|
|
print(f" 수익으로 어깨매도된 건(유지): {len(would_apply)}건")
|
|
if would_skip:
|
|
skip_pnl = sum(r["realized_pnl"] for r in would_skip)
|
|
print(f" 제외되는 건들의 실현손익 합계: {skip_pnl:+,.0f}원 (이 손실들은 다른 로직으로 나갈 때까지 보유됨)")
|
|
print()
|
|
|
|
|
|
def backtest_shoulder_rule_high_above_entry(rows, min_high_pct=0.01):
|
|
"""
|
|
개선안 B: 고점이 매수가 대비 min_high_pct 이상일 때만 어깨매도 적용.
|
|
DB에는 고점이 없으므로, "매도가가 매수가보다 높았던 건" = 수익 매도만 해당.
|
|
(실제 고점은 매도가보다 높았을 수 있음. 보수적으로 수익 건만 적용된 것처럼 집계.)
|
|
"""
|
|
if not rows:
|
|
return
|
|
# 수익 매도 = 매도가 > 매수가 → 고점은 매수가 이상이었을 것
|
|
above_entry = [r for r in rows if r["sell_price"] > r["buy_price"]]
|
|
below_entry = [r for r in rows if r["sell_price"] <= r["buy_price"]]
|
|
|
|
print("=" * 60)
|
|
print(f"🔬 백테스트 3: 개선안 B — 고점이 매수가+{min_high_pct*100:.0f}% 이상일 때만 어깨매도")
|
|
print("=" * 60)
|
|
print(" (DB에 고점 미저장으로, 매도가>매수가인 수익 건만 '고점이 매수가 위'로 간주)")
|
|
print(f" 적용될 건(수익 매도): {len(above_entry)}건")
|
|
print(f" 제외될 건(손실 매도): {len(below_entry)}건")
|
|
if below_entry:
|
|
pnl_excluded = sum(r["realized_pnl"] for r in below_entry)
|
|
print(f" 제외 건 실현손익 합계: {pnl_excluded:+,.0f}원")
|
|
print()
|
|
|
|
|
|
def main():
|
|
print("\n")
|
|
rows = report_shoulder_sell_analysis()
|
|
backtest_shoulder_rule_current(rows)
|
|
backtest_shoulder_rule_profit_only(rows)
|
|
backtest_shoulder_rule_high_above_entry(rows, min_high_pct=0.01)
|
|
print("=" * 60)
|
|
print("✅ 백테스트 완료. 개선안 반영 시 kis_short_ver2.py 매도 로직에서")
|
|
print(" 어깨매도 조건에 profit_pct > 0 또는 max_price >= buy_price * (1 + X) 추가 검토.")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|