276 lines
11 KiB
Python
276 lines
11 KiB
Python
# analyze_trades.py
|
||
# -----------------------------------------------------------------------------
|
||
# 거래 데이터 상세 분석 스크립트
|
||
# 목적: 승률, 손익 패턴, 최적화 포인트 발견
|
||
# -----------------------------------------------------------------------------
|
||
|
||
import re
|
||
from collections import defaultdict
|
||
from typing import List, Dict, Tuple
|
||
import statistics
|
||
|
||
def parse_backtest_terminal_output() -> List[Dict]:
|
||
"""터미널 출력에서 수작업 데이터 추출 (임시)"""
|
||
# 현재 13.9% 결과 요약 (백테스터 v3.0)
|
||
return {
|
||
'total_trades': 155,
|
||
'win_rate': 41.94,
|
||
'total_return': 13.90,
|
||
'cagr': 6.74,
|
||
'mdd': -15.57,
|
||
'sharpe': 0.57,
|
||
'exit_reasons': {
|
||
'TRAILING_STOP_10_30PCT': {'count': 57, 'avg': 9.48},
|
||
'STOP_LOSS_5PCT': {'count': 50, 'avg': -7.55},
|
||
'TRAILING_STOP_LT10PCT': {'count': 39, 'avg': -3.23},
|
||
'PROFIT_TAKE_FULL': {'count': 9, 'avg': 15.17}
|
||
},
|
||
'avg_win': 10.46,
|
||
'avg_loss': -5.62,
|
||
'avg_profit': 1.12
|
||
}
|
||
|
||
|
||
def parse_backtest_log(log_file: str = "current_log.txt") -> List[Dict]:
|
||
"""백테스트 로그 파싱하여 거래 데이터 추출"""
|
||
trades = []
|
||
|
||
try:
|
||
# UTF-16 LE with BOM (FF FE)
|
||
with open(log_file, 'r', encoding='utf-16') as f:
|
||
content = f.read()
|
||
except:
|
||
# 로그 파일이 없으면 터미널 출력 기반 요약 사용
|
||
return []
|
||
|
||
# 매도 로그 패턴: SELL (사유): 종목 @ 가격 / 주수 (수익률%)
|
||
sell_pattern = r'SELL\s+\(([^)]+)\):\s+([^(]+)\([^)]+\)\s+@\s+[\d,]+\s+/\s+[\d,]+주\s+\(([-+]?\d+\.\d+)%\)'
|
||
|
||
for match in re.finditer(sell_pattern, content):
|
||
reason, ticker_name, profit_pct = match.groups()
|
||
trades.append({
|
||
'ticker': ticker_name.strip(),
|
||
'profit_pct': float(profit_pct),
|
||
'exit_reason': reason,
|
||
'is_win': float(profit_pct) > 0
|
||
})
|
||
|
||
return trades
|
||
|
||
|
||
def analyze_by_exit_reason(trades: List[Dict]) -> Dict:
|
||
"""매도 사유별 승률 및 평균 수익률 분석"""
|
||
by_reason = defaultdict(list)
|
||
|
||
for trade in trades:
|
||
by_reason[trade['exit_reason']].append(trade['profit_pct'])
|
||
|
||
analysis = {}
|
||
for reason, profits in by_reason.items():
|
||
wins = [p for p in profits if p > 0]
|
||
losses = [p for p in profits if p <= 0]
|
||
|
||
analysis[reason] = {
|
||
'count': len(profits),
|
||
'win_rate': len(wins) / len(profits) * 100 if profits else 0,
|
||
'avg_profit': statistics.mean(profits),
|
||
'avg_win': statistics.mean(wins) if wins else 0,
|
||
'avg_loss': statistics.mean(losses) if losses else 0,
|
||
'total_profit': sum(profits)
|
||
}
|
||
|
||
return analysis
|
||
|
||
|
||
def analyze_by_ticker(trades: List[Dict], top_n: int = 20) -> Dict:
|
||
"""종목별 승률 및 수익률 분석 (상위 N개)"""
|
||
by_ticker = defaultdict(list)
|
||
|
||
for trade in trades:
|
||
by_ticker[trade['ticker']].append(trade['profit_pct'])
|
||
|
||
# 거래 횟수 기준 정렬
|
||
sorted_tickers = sorted(by_ticker.items(), key=lambda x: len(x[1]), reverse=True)[:top_n]
|
||
|
||
analysis = {}
|
||
for ticker, profits in sorted_tickers:
|
||
wins = [p for p in profits if p > 0]
|
||
|
||
analysis[ticker] = {
|
||
'trade_count': len(profits),
|
||
'win_rate': len(wins) / len(profits) * 100 if profits else 0,
|
||
'avg_profit': statistics.mean(profits),
|
||
'total_profit': sum(profits)
|
||
}
|
||
|
||
return analysis
|
||
|
||
|
||
def identify_optimization_opportunities(trades: List[Dict]) -> Dict:
|
||
"""최적화 기회 식별"""
|
||
|
||
# 1. 큰 손실 거래 분석 (손실률 -10% 이상)
|
||
big_losses = [t for t in trades if t['profit_pct'] < -10]
|
||
|
||
# 2. 조기 익절 후 더 올랐을 가능성 (PROFIT_TAKE_FULL이 너무 빠른지)
|
||
early_exits = [t for t in trades if t['exit_reason'] == 'PROFIT_TAKE_FULL']
|
||
|
||
# 3. 트레일링 스톱 손실 (더 늦게 팔았다면 이익?)
|
||
trailing_losses = [t for t in trades if t['exit_reason'].startswith('TRAILING_STOP') and t['profit_pct'] < 0]
|
||
|
||
return {
|
||
'big_losses': {
|
||
'count': len(big_losses),
|
||
'avg_loss': statistics.mean([t['profit_pct'] for t in big_losses]) if big_losses else 0,
|
||
'tickers': [t['ticker'] for t in big_losses]
|
||
},
|
||
'early_profit_takes': {
|
||
'count': len(early_exits),
|
||
'avg_profit': statistics.mean([t['profit_pct'] for t in early_exits]) if early_exits else 0
|
||
},
|
||
'trailing_stop_losses': {
|
||
'count': len(trailing_losses),
|
||
'avg_loss': statistics.mean([t['profit_pct'] for t in trailing_losses]) if trailing_losses else 0
|
||
}
|
||
}
|
||
|
||
|
||
def print_analysis_report(trades: List[Dict]):
|
||
"""분석 리포트 출력"""
|
||
|
||
print("=" * 80)
|
||
print("📊 백테스트 거래 상세 분석 리포트")
|
||
print("=" * 80)
|
||
|
||
# 전체 통계
|
||
total_trades = len(trades)
|
||
wins = [t for t in trades if t['is_win']]
|
||
win_rate = len(wins) / total_trades * 100 if total_trades > 0 else 0
|
||
avg_profit = statistics.mean([t['profit_pct'] for t in trades])
|
||
|
||
print(f"\n📈 전체 통계")
|
||
print(f" - 총 거래: {total_trades}건")
|
||
print(f" - 승률: {win_rate:.2f}%")
|
||
print(f" - 평균 수익률: {avg_profit:.2f}%")
|
||
print(f" - 평균 익절: {statistics.mean([t['profit_pct'] for t in wins]):.2f}%" if wins else " - 평균 익절: N/A")
|
||
print(f" - 평균 손절: {statistics.mean([t['profit_pct'] for t in trades if not t['is_win']]):.2f}%" if any(not t['is_win'] for t in trades) else "")
|
||
|
||
# 매도 사유별 분석
|
||
print(f"\n🎯 매도 사유별 분석")
|
||
by_reason = analyze_by_exit_reason(trades)
|
||
for reason, stats in sorted(by_reason.items(), key=lambda x: x[1]['count'], reverse=True):
|
||
print(f" [{reason}]")
|
||
print(f" 거래: {stats['count']}건 | 승률: {stats['win_rate']:.1f}% | 평균: {stats['avg_profit']:.2f}%")
|
||
print(f" 평균 익절: {stats['avg_win']:.2f}% | 평균 손절: {stats['avg_loss']:.2f}%")
|
||
|
||
# 종목별 분석 (Top 10)
|
||
print(f"\n🏆 거래 빈도 상위 종목 (Top 10)")
|
||
by_ticker = analyze_by_ticker(trades, top_n=10)
|
||
for ticker, stats in by_ticker.items():
|
||
print(f" {ticker}: {stats['trade_count']}건 | 승률: {stats['win_rate']:.1f}% | 평균: {stats['avg_profit']:.2f}% | 누적: {stats['total_profit']:.2f}%")
|
||
|
||
# 최적화 기회
|
||
print(f"\n🔍 최적화 기회 분석")
|
||
opps = identify_optimization_opportunities(trades)
|
||
|
||
print(f"\n ⚠️ 큰 손실 거래 (-10% 이상)")
|
||
print(f" 건수: {opps['big_losses']['count']}건 | 평균 손실: {opps['big_losses']['avg_loss']:.2f}%")
|
||
if opps['big_losses']['count'] > 0:
|
||
print(f" 문제 종목: {', '.join(set(opps['big_losses']['tickers'][:5]))}")
|
||
|
||
print(f"\n 💰 익절 (PROFIT_TAKE_FULL)")
|
||
print(f" 건수: {opps['early_profit_takes']['count']}건 | 평균 익절: {opps['early_profit_takes']['avg_profit']:.2f}%")
|
||
print(f" → 현재 15% 익절 설정이 적절한지 검토")
|
||
|
||
print(f"\n 📉 트레일링 스톱 손실")
|
||
print(f" 건수: {opps['trailing_stop_losses']['count']}건 | 평균 손실: {opps['trailing_stop_losses']['avg_loss']:.2f}%")
|
||
print(f" → 트레일링 스톱 비율 조정 검토 (현재: 10%/10%/15%)")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("💡 최적화 제안")
|
||
print("=" * 80)
|
||
|
||
# 자동 제안 생성
|
||
suggestions = []
|
||
|
||
if opps['big_losses']['count'] > total_trades * 0.2:
|
||
suggestions.append("1. 큰 손실 거래가 20% 이상 → 손절 로직 강화 필요")
|
||
|
||
if opps['early_profit_takes']['avg_profit'] < 20:
|
||
suggestions.append("2. 평균 익절이 20% 미만 → SELL_PROFIT_TAKE_PCT를 20%로 상향 테스트")
|
||
|
||
if opps['trailing_stop_losses']['count'] > 30:
|
||
suggestions.append("3. 트레일링 스톱 손실 많음 → 비율을 12-15%로 완화 테스트")
|
||
|
||
if win_rate < 45:
|
||
suggestions.append(f"4. 승률 {win_rate:.1f}% → 매수 필터 강화 (이격도/거래량 조건 추가)")
|
||
|
||
if not suggestions:
|
||
suggestions.append("✅ 현재 설정이 균형잡혀 있습니다. 미세 조정 권장.")
|
||
|
||
for suggestion in suggestions:
|
||
print(f" {suggestion}")
|
||
|
||
print("\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 터미널 출력 기반 요약 분석
|
||
data = parse_backtest_terminal_output()
|
||
|
||
print("=" * 80)
|
||
print("📊 백테스트 거래 상세 분석 리포트 (v3.0 - 13.9% 결과)")
|
||
print("=" * 80)
|
||
|
||
print(f"\n📈 전체 통계")
|
||
print(f" - 총 거래: {data['total_trades']}건")
|
||
print(f" - 승률: {data['win_rate']:.2f}%")
|
||
print(f" - 총 수익률: {data['total_return']:.2f}%")
|
||
print(f" - CAGR: {data['cagr']:.2f}%")
|
||
print(f" - MDD: {data['mdd']:.2f}%")
|
||
print(f" - 평균 수익률: {data['avg_profit']:.2f}%")
|
||
print(f" - 평균 익절: {data['avg_win']:.2f}%")
|
||
print(f" - 평균 손절: {data['avg_loss']:.2f}%")
|
||
|
||
print(f"\n🎯 매도 사유별 분석")
|
||
for reason, stats in data['exit_reasons'].items():
|
||
print(f" [{reason}]")
|
||
print(f" 거래: {stats['count']}건 | 평균: {stats['avg']:.2f}%")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("💡 최적화 제안 (현재 시스템 v3.0)")
|
||
print("=" * 80)
|
||
|
||
print(f"\n 1️⃣ **트레일링 스톱 완화 테스트** ⭐ 최우선")
|
||
print(f" - 현재: 10%/10%/15% → 제안: 12%/12%/18%")
|
||
print(f" - TRAILING_STOP_LT10PCT 39건(평균 -3.23%) 개선 가능")
|
||
print(f" - 예상 효과: 조기 손절 방지, 승률 45% 목표")
|
||
|
||
print(f"\n 2️⃣ **익절 기준 상향 테스트**")
|
||
print(f" - 현재: 15% 익절 → 제안: 20% 익절")
|
||
print(f" - PROFIT_TAKE_FULL 9건(평균 15.17%) 더 높은 수익 포착")
|
||
print(f" - 예상 효과: 평균 익절 10.46% → 12% 목표")
|
||
|
||
print(f"\n 3️⃣ **손절 로직 강화**")
|
||
print(f" - STOP_LOSS_5PCT 50건(평균 -7.55%) = 가장 큰 손실 원인")
|
||
print(f" - 제안: USE_TECHNICAL_STOPLOSS 단독 사용 (고정 7% 손절 제거)")
|
||
print(f" - 예상 효과: 평균 손실 -7.55% → -6% 목표")
|
||
|
||
print(f"\n 4️⃣ **매수 필터 강화 (승률 45% 목표)**")
|
||
print(f" - 현재 승률 41.94% → 목표 45%+")
|
||
print(f" - 제안: USE_DISPARITY_FILTER = True (이격도 필터 복원)")
|
||
print(f" - 또는: USE_STRONG_BREAKTHROUGH_FILTER = True")
|
||
|
||
print(f"\n 📊 예상 개선 시나리오:")
|
||
print(f" - 보수적 목표: 13.9% → 16~17% (+2~3%p)")
|
||
print(f" - 공격적 목표: 13.9% → 18~20% (+4~6%p)")
|
||
print(f" - 달성 방법: 트레일링 스톱 완화 + 익절 상향 조합")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("🚀 다음 단계")
|
||
print("=" * 80)
|
||
print(" 1. config.py 수정: SELL_TRAILING_STOP_*_PCT 값 조정 (10→12, 15→18)")
|
||
print(" 2. python backtester.py 실행하여 결과 비교")
|
||
print(" 3. 개선 시 추가 파라미터 미세 조정")
|
||
print("\n")
|