cyb50-quant/cat-fly/auto_report.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 自动化交易报告系统 (独立版)
功能：
1. 获取近2个月数据
2. 运行策略回测
3. 生成详细报告
4. 发送邮件通知

执行频率：A股开盘时间每半小时（9:30-11:30, 13:00-15:00）
"""

import pandas as pd
import numpy as np
import akshare as ak
import warnings
import os
import smtplib
import ssl
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
warnings.filterwarnings('ignore')

# ==================== 邮件配置 ====================
# 使用本地Postfix SMTP服务器发送
EMAIL_CONFIG = {
    "smtp_server": "localhost",         # 本地Postfix服务器
    "smtp_port": 25,                    # SMTP端口
    "sender_email": "catfly@openclaw.local",  # 发件人邮箱
    "sender_password": "",              # 本地SMTP无需密码
    "receiver_email": "380880504@qq.com"      # 收件人邮箱
}

def send_email(subject, html_content, text_content=""):
    """发送邮件 - 使用本地Postfix"""
    try:
        msg = MIMEMultipart('alternative')
        msg['Subject'] = Header(subject, 'utf-8')
        msg['From'] = EMAIL_CONFIG['sender_email']
        msg['To'] = EMAIL_CONFIG['receiver_email']
        
        # 纯文本版本
        text_part = MIMEText(text_content, 'plain', 'utf-8')
        msg.attach(text_part)
        
        # HTML版本
        html_part = MIMEText(html_content, 'html', 'utf-8')
        msg.attach(html_part)
        
        # 发送邮件 - 本地Postfix无需SSL和认证
        with smtplib.SMTP(EMAIL_CONFIG['smtp_server'], EMAIL_CONFIG['smtp_port']) as server:
            server.sendmail(
                EMAIL_CONFIG['sender_email'],
                EMAIL_CONFIG['receiver_email'],
                msg.as_string()
            )
        print(f"✅ 邮件发送成功: {subject}")
        return True
    except Exception as e:
        print(f"❌ 邮件发送失败: {e}")
        print(f"   请检查EMAIL_CONFIG配置是否正确")
        return False


# ==================== 数据获取 ====================
class DataFetcher:
    """数据获取类 - 使用实时在线数据"""
    
    @staticmethod
    def fetch_recent_2months():
        """获取近2个月数据 - 使用实时在线数据（东方财富+新浪财经双数据源）"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=70)  # 2个月+10天缓冲
        
        print(f"获取数据: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}")
        
        # 尝试东方财富数据源
        df = DataFetcher._fetch_eastmoney_data(start_date, end_date)
        if df is not None:
            return df
        
        # 东方财富失败，尝试新浪财经
        print("⚠️ 东方财富数据源失败，尝试新浪财经...")
        df = DataFetcher._fetch_sina_data(start_date, end_date)
        if df is not None:
            return df
        
        # 所有数据源都失败
        raise Exception("无法获取实时数据，东方财富和新浪财经均失败。请检查网络连接或稍后重试。")
    
    @staticmethod
    def _fetch_eastmoney_data(start_date, end_date):
        """从东方财富获取数据"""
        try:
            print("[数据源1] 正在使用东方财富30分钟K线接口...")
            
            # 使用东方财富接口获取30分钟K线
            df = ak.index_zh_a_hist_min_em(symbol="399673", period="30")
            
            if df is not None and not df.empty and len(df) >= 50:
                # 标准化列名
                df = df.rename(columns={
                    '时间': 'datetime',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume'
                })
                
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                
                # 只保留最近2个月的数据用于回测
                backtest_start = end_date - timedelta(days=60)
                df_backtest = df[df.index >= backtest_start]
                
                print(f"✅ 东方财富数据获取成功: 共{len(df_backtest)}条30分钟K线")
                print(f"   数据区间: {df_backtest.index[0]} 至 {df_backtest.index[-1]}")
                
                # 检查数据时效性
                latest_time = df_backtest.index[-1]
                time_delay = end_date - latest_time
                print(f"   数据延迟: {time_delay}")
                
                return df_backtest
            else:
                print(f"⚠️ 东方财富数据不足: {len(df) if df is not None else 0}条")
                return None
                
        except Exception as e:
            print(f"❌ 东方财富数据源失败: {e}")
            return None
    
    @staticmethod
    def _fetch_sina_data(start_date, end_date):
        """从新浪财经获取数据"""
        try:
            print("[数据源2] 正在使用新浪财经30分钟K线接口...")
            
            import requests
            import json
            import re
            
            symbol = "sz399673"
            url = f"https://quotes.sina.cn/cn/api/jsonp_v2.php/var_{symbol}_30_/CN_MarketDataService.getKLineData?symbol={symbol}&scale=30&ma=no&datalen=1023"
            
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
                'Referer': 'https://finance.sina.com.cn/'
            }
            
            response = requests.get(url, headers=headers, timeout=15)
            response_text = response.text
            
            # 解析JSONP响应
            json_start = response_text.find('[')
            json_end = response_text.rfind(']') + 1
            if json_start >= 0 and json_end > json_start:
                json_str = response_text[json_start:json_end]
            else:
                raise Exception("无法解析JSONP响应")
            
            data_dict = json.loads(json_str)
            
            if data_dict and isinstance(data_dict, list) and len(data_dict) > 0:
                data_list = []
                for item in data_dict:
                    try:
                        data_list.append({
                            'datetime': item.get('day'),
                            'open': float(item.get('open', 0)),
                            'high': float(item.get('high', 0)),
                            'low': float(item.get('low', 0)),
                            'close': float(item.get('close', 0)),
                            'volume': float(item.get('volume', 0))
                        })
                    except Exception:
                        continue
                
                if data_list:
                    df = pd.DataFrame(data_list)
                    df['datetime'] = pd.to_datetime(df['datetime'])
                    df = df.set_index('datetime').sort_index()
                    
                    # 只保留最近2个月的数据
                    backtest_start = end_date - timedelta(days=60)
                    df_backtest = df[df.index >= backtest_start]
                    
                    print(f"✅ 新浪财经数据获取成功: 共{len(df_backtest)}条30分钟K线")
                    print(f"   数据区间: {df_backtest.index[0]} 至 {df_backtest.index[-1]}")
                    
                    return df_backtest
                else:
                    print("❌ 新浪财经数据解析失败")
                    return None
            else:
                print("❌ 新浪财经返回数据格式错误")
                return None
                
        except Exception as e:
            print(f"❌ 新浪财经数据源失败: {e}")
            return None


# ==================== 策略类 ====================
class CatFlyStrategy:
    """cat-fly策略简化版 - 基于30分钟K线"""
    
    def __init__(self, config=None):
        self.config = config or {
            'initial_capital': 1000000,
            'position_size_pct': 1.0,
            'stop_loss_pct': 0.008,
            'take_profit_pct': 0.02,
            'max_hold_bars': 16,
            'min_signal_strength': 3
        }
        self.initial_capital = self.config['initial_capital']
    
    def calculate_indicators(self, df):
        """计算技术指标"""
        # RSI
        delta = df['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))
        
        # 移动平均线
        df['MA5'] = df['close'].rolling(5).mean()
        df['MA20'] = df['close'].rolling(20).mean()
        df['MA60'] = df['close'].rolling(60).mean()
        
        # 布林带
        df['BB_middle'] = df['close'].rolling(20).mean()
        bb_std = df['close'].rolling(20).std()
        df['BB_upper'] = df['BB_middle'] + 2 * bb_std
        df['BB_lower'] = df['BB_middle'] - 2 * bb_std
        
        # MACD
        ema12 = df['close'].ewm(span=12).mean()
        ema26 = df['close'].ewm(span=26).mean()
        df['MACD'] = ema12 - ema26
        df['MACD_signal'] = df['MACD'].ewm(span=9).mean()
        
        return df
    
    def generate_signals(self, df):
        """生成交易信号"""
        df = self.calculate_indicators(df)
        df['signal'] = 0
        df['signal_strength'] = 0
        
        for i in range(60, len(df)):
            row = df.iloc[i]
            strength = 0
            
            # RSI超卖/超买
            if row['RSI'] < 30:
                strength += 1
            elif row['RSI'] > 70:
                strength -= 1
            
            # 均线多头排列/空头排列
            if row['close'] > row['MA5'] > row['MA20']:
                strength += 1
            elif row['close'] < row['MA5'] < row['MA20']:
                strength -= 1
            
            # 布林带
            if row['close'] < row['BB_lower']:
                strength += 1
            elif row['close'] > row['BB_upper']:
                strength -= 1
            
            # MACD金叉/死叉
            if i > 0:
                prev_macd = df['MACD'].iloc[i-1]
                prev_signal = df['MACD_signal'].iloc[i-1]
                curr_macd = row['MACD']
                curr_signal_line = row['MACD_signal']
                
                if prev_macd < prev_signal and curr_macd > curr_signal_line:
                    strength += 1
                elif prev_macd > prev_signal and curr_macd < curr_signal_line:
                    strength -= 1
            
            df.iloc[i, df.columns.get_loc('signal_strength')] = strength
            
            # 生成交易信号
            if strength >= self.config['min_signal_strength']:
                df.iloc[i, df.columns.get_loc('signal')] = 1  # 做多
            elif strength <= -self.config['min_signal_strength']:
                df.iloc[i, df.columns.get_loc('signal')] = -1  # 做空
        
        return df
    
    def backtest(self, df):
        """回测"""
        df = self.generate_signals(df)
        
        trades = []
        capital = self.initial_capital
        position = 0
        entry_price = 0
        entry_time = None
        holding_bars = 0
        
        for i in range(60, len(df)):
            current_bar = df.iloc[i]
            price = current_bar['close']
            current_time = current_bar.name
            
            # 无持仓时检查开仓信号
            if position == 0:
                if current_bar['signal'] == 1:  # 做多
                    position_size = int(capital * self.config['position_size_pct'] / price)
                    if position_size > 0:
                        position = position_size
                        entry_price = price
                        entry_time = current_time
                        holding_bars = 0
                        
                elif current_bar['signal'] == -1:  # 做空
                    position_size = int(capital * self.config['position_size_pct'] / price)
                    if position_size > 0:
                        position = -position_size
                        entry_price = price
                        entry_time = current_time
                        holding_bars = 0
            
            # 有持仓时检查平仓
            else:
                holding_bars += 1
                exit_signal = False
                exit_reason = ""
                
                if position > 0:  # 做多持仓
                    if price <= entry_price * (1 - self.config['stop_loss_pct']):
                        exit_signal = True
                        exit_reason = "止损"
                    elif price >= entry_price * (1 + self.config['take_profit_pct']):
                        exit_signal = True
                        exit_reason = "止盈"
                    elif holding_bars >= self.config['max_hold_bars']:
                        exit_signal = True
                        exit_reason = "时间止损"
                    elif current_bar['RSI'] > 70:
                        exit_signal = True
                        exit_reason = "信号消失(RSI超买)"
                
                else:  # 做空持仓
                    if price >= entry_price * (1 + self.config['stop_loss_pct']):
                        exit_signal = True
                        exit_reason = "止损"
                    elif price <= entry_price * (1 - self.config['take_profit_pct']):
                        exit_signal = True
                        exit_reason = "止盈"
                    elif holding_bars >= self.config['max_hold_bars']:
                        exit_signal = True
                        exit_reason = "时间止损"
                    elif current_bar['RSI'] < 30:
                        exit_signal = True
                        exit_reason = "信号消失(RSI超卖)"
                
                # 执行平仓
                if exit_signal:
                    if position > 0:
                        pnl = (price - entry_price) * position
                        pnl_pct = (price - entry_price) / entry_price * 100
                    else:
                        pnl = (entry_price - price) * abs(position)
                        pnl_pct = (entry_price - price) / entry_price * 100
                    
                    capital += pnl
                    
                    trades.append({
                        '方向': '做多' if position > 0 else '做空',
                        '开仓时间': entry_time,
                        '平仓时间': current_time,
                        '开仓价': entry_price,
                        '平仓价': price,
                        '持仓数量': abs(position),
                        '盈亏金额': pnl,
                        '盈亏百分比': pnl_pct,
                        '退出原因': exit_reason,
                        '持仓周期': holding_bars,
                        '平仓后资金': capital
                    })
                    
                    position = 0
                    entry_price = 0
                    entry_time = None
                    holding_bars = 0
        
        return df, pd.DataFrame(trades), capital


# ==================== 报告生成 ====================
def generate_report(trades_df, final_capital, initial_capital=1000000):
    """生成详细报告"""
    
    if len(trades_df) == 0:
        html = "<html><body><h1>创业板50交易报告</h1><p>近2个月无交易信号</p></body></html>"
        text = "近2个月无交易信号"
        return html, text
    
    total_return = (final_capital - initial_capital) / initial_capital * 100
    total_trades = len(trades_df)
    winning_trades = trades_df[trades_df['盈亏金额'] > 0]
    losing_trades = trades_df[trades_df['盈亏金额'] < 0]
    
    win_rate = len(winning_trades) / total_trades * 100 if total_trades > 0 else 0
    avg_profit = winning_trades['盈亏金额'].mean() if len(winning_trades) > 0 else 0
    avg_loss = losing_trades['盈亏金额'].mean() if len(losing_trades) > 0 else 0
    
    total_profit = winning_trades['盈亏金额'].sum() if len(winning_trades) > 0 else 0
    total_loss = abs(losing_trades['盈亏金额'].sum()) if len(losing_trades) > 0 else 0
    profit_factor = total_profit / total_loss if total_loss > 0 else 0
    
    max_profit = trades_df['盈亏金额'].max()
    max_loss = trades_df['盈亏金额'].min()
    avg_hold_time = trades_df['持仓周期'].mean()
    
    long_trades = trades_df[trades_df['方向'] == '做多']
    short_trades = trades_df[trades_df['方向'] == '做空']
    exit_reasons = trades_df['退出原因'].value_counts()
    
    # 生成HTML报告
    html = f"""
    <html>
    <head>
        <style>
            body {{ font-family: Arial, sans-serif; margin: 20px; }}
            h1 {{ color: #333; border-bottom: 2px solid #007bff; padding-bottom: 10px; }}
            h2 {{ color: #555; margin-top: 30px; }}
            table {{ border-collapse: collapse; width: 100%; margin: 15px 0; font-size: 14px; }}
            th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
            th {{ background-color: #007bff; color: white; }}
            tr:nth-child(even) {{ background-color: #f2f2f2; }}
            .positive {{ color: green; font-weight: bold; }}
            .negative {{ color: red; font-weight: bold; }}
            .summary {{ background-color: #f8f9fa; padding: 15px; border-radius: 5px; margin: 15px 0; }}
        </style>
    </head>
    <body>
        <h1>🚀 创业板50指数交易报告</h1>
        <p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
        <p>数据区间: 近2个月</p>
        
        <div class="summary">
            <h2>📊 总体绩效</h2>
            <table>
                <tr><th>指标</th><th>数值</th></tr>
                <tr><td>初始资金</td><td>{initial_capital:,.0f}元</td></tr>
                <tr><td>最终资金</td><td>{final_capital:,.0f}元</td></tr>
                <tr><td>总收益率</td><td class="{'positive' if total_return >= 0 else 'negative'}">{total_return:+.2f}%</td></tr>
                <tr><td>总交易次数</td><td>{total_trades}笔</td></tr>
                <tr><td>胜率</td><td>{win_rate:.1f}%</td></tr>
                <tr><td>盈亏比</td><td>{profit_factor:.2f}</td></tr>
                <tr><td>平均持仓时间</td><td>{avg_hold_time:.1f}周期 ({avg_hold_time*0.5:.1f}小时)</td></tr>
            </table>
        </div>
        
        <h2>📈 盈亏统计</h2>
        <table>
            <tr><th>指标</th><th>数值</th></tr>
            <tr><td>总盈利</td><td class="positive">+{total_profit:,.0f}元</td></tr>
            <tr><td>总亏损</td><td class="negative">-{total_loss:,.0f}元</td></tr>
            <tr><td>平均盈利</td><td class="positive">+{avg_profit:,.0f}元</td></tr>
            <tr><td>平均亏损</td><td class="negative">{avg_loss:,.0f}元</td></tr>
            <tr><td>最大单笔盈利</td><td class="positive">+{max_profit:,.0f}元</td></tr>
            <tr><td>最大单笔亏损</td><td class="negative">{max_loss:,.0f}元</td></tr>
        </table>
        
        <h2>🔄 多空统计</h2>
        <table>
            <tr><th>方向</th><th>交易次数</th><th>胜率</th><th>总盈亏</th></tr>
            <tr>
                <td>做多</td>
                <td>{len(long_trades)}笔</td>
                <td>{(len(long_trades[long_trades['盈亏金额']>0])/len(long_trades)*100 if len(long_trades)>0 else 0):.1f}%</td>
                <td class="{'positive' if long_trades['盈亏金额'].sum() >= 0 else 'negative'}">{long_trades['盈亏金额'].sum():+,.0f}元</td>
            </tr>
            <tr>
                <td>做空</td>
                <td>{len(short_trades)}笔</td>
                <td>{(len(short_trades[short_trades['盈亏金额']>0])/len(short_trades)*100 if len(short_trades)>0 else 0):.1f}%</td>
                <td class="{'positive' if short_trades['盈亏金额'].sum() >= 0 else 'negative'}">{short_trades['盈亏金额'].sum():+,.0f}元</td>
            </tr>
        </table>
        
        <h2>🚪 退出原因分析</h2>
        <table>
            <tr><th>退出原因</th><th>次数</th><th>占比</th></tr>
    """
    
    for reason, count in exit_reasons.items():
        pct = count / total_trades * 100
        html += f"<tr><td>{reason}</td><td>{count}</td><td>{pct:.1f}%</td></tr>"
    
    html += """
        </table>
        
        <h2>📝 最近10笔交易明细</h2>
        <table>
            <tr>
                <th>方向</th>
                <th>开仓时间</th>
                <th>平仓时间</th>
                <th>开仓价</th>
                <th>平仓价</th>
                <th>盈亏金额</th>
                <th>盈亏%</th>
                <th>退出原因</th>
            </tr>
    """
    
    recent_trades = trades_df.tail(10)
    for _, trade in recent_trades.iterrows():
        pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative"
        html += f"""
            <tr>
                <td>{trade['方向']}</td>
                <td>{trade['开仓时间']}</td>
                <td>{trade['平仓时间']}</td>
                <td>{trade['开仓价']:.2f}</td>
                <td>{trade['平仓价']:.2f}</td>
                <td class="{pnl_class}">{trade['盈亏金额']:+.0f}</td>
                <td class="{pnl_class}">{trade['盈亏百分比']:+.2f}%</td>
                <td>{trade['退出原因']}</td>
            </tr>
        """
    
    html += """
        </table>
        
        <hr>
        <p style="color: #666; font-size: 12px;">
            本报告由 cat-fly 自动交易系统生成 | 策略：30分钟K线多空双向<br>
            风险提示：历史回测不代表未来表现，投资有风险，入市需谨慎。
        </p>
    </body>
    </html>
    """
    
    # 生成纯文本版本
    text = f"""
创业板50指数交易报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
数据区间: 近2个月

【总体绩效】
初始资金: {initial_capital:,.0f}元
最终资金: {final_capital:,.0f}元
总收益率: {total_return:+.2f}%
总交易次数: {total_trades}笔
胜率: {win_rate:.1f}%
盈亏比: {profit_factor:.2f}
平均持仓: {avg_hold_time*0.5:.1f}小时

【盈亏统计】
总盈利: +{total_profit:,.0f}元
总亏损: -{total_loss:,.0f}元
最大单笔盈利: +{max_profit:,.0f}元
最大单笔亏损: {max_loss:,.0f}元

【多空统计】
做多: {len(long_trades)}笔, 盈亏{long_trades['盈亏金额'].sum():+,.0f}元
做空: {len(short_trades)}笔, 盈亏{short_trades['盈亏金额'].sum():+,.0f}元

【退出原因】
{exit_reasons.to_string()}

【最近5笔交易】
{trades_df.tail(5)[['方向', '开仓时间', '平仓时间', '盈亏金额', '退出原因']].to_string(index=False)}
    """
    
    return html, text


# ==================== 主程序 ====================
def main():
    """主程序"""
    print("="*80)
    print("🚀 cat-fly 自动交易报告系统")
    print("="*80)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 检查是否在交易时间（可选）
    now = datetime.now()
    hour = now.hour
    minute = now.minute
    time_str = f"{hour:02d}:{minute:02d}"
    
    # A股交易时间检查
    is_trading_time = False
    if (9 <= hour <= 11) or (13 <= hour <= 15):
        if hour == 9 and minute < 30:
            is_trading_time = False
        elif hour == 11 and minute > 30:
            is_trading_time = False
        elif hour == 15 and minute > 0:
            is_trading_time = False
        else:
            is_trading_time = True
    
    print(f"当前时间: {time_str}")
    print(f"交易时间: {'是' if is_trading_time else '否（非交易时间也会执行）'}")
    
    # 1. 获取近2个月数据
    print("\n📊 步骤1: 获取近2个月数据...")
    df = DataFetcher.fetch_recent_2months()
    if df is None:
        print("❌ 数据获取失败，退出")
        return
    
    # 2. 运行策略
    print("\n📈 步骤2: 运行策略回测...")
    strategy = CatFlyStrategy()
    df, trades_df, final_capital = strategy.backtest(df)
    
    print(f"✅ 回测完成: 共{len(trades_df)}笔交易")
    print(f"   最终资金: {final_capital:,.0f}元")
    print(f"   收益率: {(final_capital/1000000-1)*100:+.2f}%")
    
    # 3. 生成报告
    print("\n📝 步骤3: 生成报告...")
    html_report, text_report = generate_report(trades_df, final_capital)
    
    # 4. 发送邮件
    print("\n📧 步骤4: 发送邮件...")
    subject = f"🚀 创业板50交易报告 {datetime.now().strftime('%m-%d %H:%M')} | 收益{(final_capital/1000000-1)*100:+.2f}%"
    send_email(subject, html_report, text_report)
    
    print("\n✅ 全部完成!")
    print("="*80)


if __name__ == "__main__":
    main()