kis_trader/remove/legacy_root/auto_ai_reporter.py

#!/usr/bin/env python3
"""
자동 AI 분석 보고서
- 매일 13:00에 최근 10건 거래 분석
- AI가 문제점 진단 및 .env 수정 권장
"""

import os
import sys
import time
import json
import sqlite3
import datetime
import warnings
import requests
from dotenv import load_dotenv

# .env 로드
current_dir = os.path.dirname(os.path.abspath(__file__))
env_file = os.path.join(current_dir, '.env')
load_dotenv(env_file)

# Gemini deprecated 경고 억제
warnings.filterwarnings("ignore", message=".*google.generativeai.*")

# Gemini API
try:
    import google.generativeai as genai
    GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
    if GEMINI_API_KEY:
        genai.configure(api_key=GEMINI_API_KEY)
        model = genai.GenerativeModel('gemini-2.5-flash')
    else:
        model = None
except Exception as e:
    print(f"❌ Gemini 초기화 실패: {e}")
    model = None

# Mattermost (Ver2와 동일: mm_config.json + MM_BOT_TOKEN_)
MM_SERVER_URL = os.getenv("MM_SERVER_URL", "https://mattermost.hoonfam.org")
MM_TOKEN = os.environ.get("MM_BOT_TOKEN_", os.getenv("MATTERMOST_TOKEN", "")).strip()
MM_CONFIG_FILE = os.path.join(current_dir, "mm_config.json")
MM_CHANNEL = os.getenv("MATTERMOST_CHANNEL", "stock")

def _load_mm_channels():
    try:
        if os.path.exists(MM_CONFIG_FILE):
            with open(MM_CONFIG_FILE, 'r', encoding='utf-8') as f:
                return json.load(f).get("channels", {})
    except Exception:
        pass
    return {}

def send_mm(message):
    """Mattermost 전송"""
    channels = _load_mm_channels()
    channel_id = channels.get(MM_CHANNEL)
    if not channel_id or not MM_TOKEN:
        print("❌ MM 설정 없음 (mm_config.json 또는 MM_BOT_TOKEN_)")
        return False
    api_url = f"{MM_SERVER_URL.rstrip('/')}/api/v4/posts"
    headers = {"Authorization": f"Bearer {MM_TOKEN}", "Content-Type": "application/json"}
    payload = {"channel_id": channel_id, "message": message}
    try:
        r = requests.post(api_url, headers=headers, json=payload, timeout=5)
        r.raise_for_status()
        return True
    except Exception as e:
        print(f"❌ MM 전송 에러: {e}")
        return False

def get_recent_trades(limit=10):
    """최근 거래 N건 조회"""
    db_path = os.path.join(os.path.dirname(__file__), 'quant_bot.db')
    conn = sqlite3.connect(db_path)
    
    cursor = conn.execute(f"""
        SELECT 
            code, name, 
            buy_price, sell_price, qty,
            profit_rate, realized_pnl,
            strategy, sell_reason,
            buy_date, sell_date, hold_minutes
        FROM trade_history
        ORDER BY id DESC
        LIMIT {limit}
    """)
    
    trades = []
    for row in cursor.fetchall():
        trades.append({
            'code': row[0],
            'name': row[1],
            'buy_price': row[2],
            'sell_price': row[3],
            'qty': row[4],
            'profit_rate': row[5],
            'realized_pnl': row[6],
            'strategy': row[7],
            'sell_reason': row[8],
            'buy_date': row[9],
            'sell_date': row[10],
            'hold_minutes': row[11]
        })
    
    conn.close()
    return trades

def get_trade_summary(trades):
    """거래 통계 요약"""
    if not trades:
        return "거래 없음"
    
    total = len(trades)
    wins = sum(1 for t in trades if t['profit_rate'] > 0)
    losses = total - wins
    win_rate = wins / total * 100 if total > 0 else 0
    
    avg_profit = sum(t['profit_rate'] for t in trades) / total
    total_pnl = sum(t['realized_pnl'] for t in trades)
    
    avg_hold = sum(t['hold_minutes'] for t in trades) / total
    
    return f"""
📊 최근 {total}건 거래 통계
- 승률: {win_rate:.1f}% ({wins}승 {losses}패)
- 평균 수익률: {avg_profit:.2f}%
- 총 손익: {total_pnl:,.0f}원
- 평균 보유: {avg_hold:.0f}분
"""

def analyze_with_ai(trades):
    """AI로 거래 분석 및 권장사항"""
    if not model:
        return "❌ AI 분석 불가 (Gemini API 키 없음)"
    
    trades_text = ""
    for i, t in enumerate(trades, 1):
        trades_text += f"""
[거래 {i}] {t['name']} ({t['strategy']})
- 매수: {t['buy_price']:,.0f}원 × {t['qty']}주
- 매도: {t['sell_price']:,.0f}원
- 손익: {t['profit_rate']:+.2f}% ({t['realized_pnl']:,.0f}원)
- 보유: {t['hold_minutes']}분
- 사유: {t['sell_reason']}
"""
    
    prompt = f"""당신은 퀀트 트레이딩 전문가입니다.

다음은 최근 {len(trades)}건의 거래 내역입니다:

{trades_text}

{get_trade_summary(trades)}

**당신의 임무:**
1. 문제점 3가지 진단 (구체적으로)
2. .env 수정 권장사항 (변수명=값 형식)
3. 예상 효과

**출력 형식:**
## 🔍 문제점
1. [구체적 문제 1]
2. [구체적 문제 2]
3. [구체적 문제 3]

## 💡 권장 수정사항
```
RSI_OVERHEAT_THRESHOLD=XX
HIGH_PRICE_CHASE_THRESHOLD=X.XX
STOP_LOSS_PCT=-X.XX
...
```

## 📈 예상 효과
- [효과 1]
- [효과 2]

**간결하고 명확하게 답변하세요.**
"""
    
    try:
        response = model.generate_content(prompt)
        return response.text
    except Exception as e:
        return f"❌ AI 분석 실패: {e}"

def send_daily_report():
    """13시 정기 보고서"""
    print(f"\n{'='*80}")
    print(f"🤖 자동 AI 분석 시작: {datetime.datetime.now()}")
    print(f"{'='*80}\n")
    
    trades = get_recent_trades(10)
    
    if not trades:
        print("❌ 거래 내역 없음")
        return
    
    print("🧠 AI 분석 중...")
    analysis = analyze_with_ai(trades)
    
    summary = get_trade_summary(trades)
    
    message = f"""🤖 **[13시 AI 자동 분석]**

{summary}

{analysis}

---
💬 명령어 사용법:
- `!env 보기` - 현재 설정 확인
- `!env RSI_OVERHEAT_THRESHOLD=75` - 설정 변경
- `!ai 왜 승률이 낮아?` - AI 질문
"""
    
    if send_mm(message):
        print("✅ 보고서 전송 완료")
    else:
        print("❌ 전송 실패")

def main():
    """메인 루프"""
    print("🤖 자동 AI 분석 보고서 시작")
    print(f"- 보고 시간: 매일 13:00")
    print(f"- 분석 대상: 최근 10건 거래")
    print(f"- AI 모델: Gemini 2.5 Flash")
    print()
    
    if not model:
        print("❌ Gemini API 키가 없습니다.")
        return
    
    reported_today = False
    main.last_date = datetime.date.today()
    
    while True:
        now = datetime.datetime.now()
        current_date = now.date()
        
        if main.last_date != current_date:
            reported_today = False
            print(f"\n📅 날짜 변경: {current_date}")
        
        main.last_date = current_date
        
        if now.hour == 13 and now.minute == 0 and not reported_today:
            send_daily_report()
            reported_today = True
            time.sleep(60)
        
        time.sleep(60)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n👋 종료")
    except Exception as e:
        print(f"\n\n❌ 에러: {e}")
        import traceback
        traceback.print_exc()