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')}")
        
        # 首先尝试在线获取
        try:
            df = ak.index_zh_a_hist_min_em(
                symbol="399673",
                period="30",
                start_date=start_date.strftime('%Y%m%d%H%M%S'),
                end_date=end_date.strftime('%Y%m%d%H%M%S')
            )
            
            if df is not None and len(df) > 0:
                df = df.rename(columns={
                    '时间': 'datetime',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume'
                })
                
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                
                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
        except Exception as e:
            print(f"⚠️ 在线数据获取失败: {e}")
            print("   尝试使用本地数据...")
        
        # 在线获取失败，使用本地数据
        return DataFetcher._load_local_data(start_date, end_date)
    
    @staticmethod
    def _load_local_data(start_date, end_date):
        """从本地文件加载数据"""
        local_file = 'SZ#399673.txt'
        
        try:
            if not os.path.exists(local_file):
                print(f"❌ 本地文件不存在: {local_file}")
                return None
            
            print(f"正在从本地文件读取数据: {local_file}")
            
            # 读取文本格式数据
            data_list = []
            encodings = ['gbk', 'gb2312', 'utf-8', 'latin-1']
            lines = None
            
            for encoding in encodings:
                try:
                    with open(local_file, 'r', encoding=encoding) as f:
                        lines = f.readlines()
                    print(f"✅ 成功使用编码: {encoding}")
                    break
                except UnicodeDecodeError:
                    continue
            
            if lines is None:
                raise ValueError("无法读取文件，尝试了多种编码格式都失败")
            
            # 跳过前两行（标题行）
            for line in lines[2:]:
                line = line.strip()
                if not line:
                    continue
                    
                parts = line.split()
                if len(parts) >= 7:
                    try:
                        # 格式: 日期 时间 开盘 最高 最低 收盘 成交量 成交额
                        date_time_str = f"{parts[0]} {parts[1]}"
                        datetime_obj = pd.to_datetime(date_time_str, format='%Y/%m/%d %H%M')
                        
                        data_list.append({
                            'datetime': datetime_obj,
                            'open': float(parts[2]),
                            'high': float(parts[3]),
                            'low': float(parts[4]),
                            'close': float(parts[5]),
                            'volume': float(parts[6])
                        })
                    except (ValueError, IndexError) as e:
                        continue
            
            if not data_list:
                raise ValueError("文本文件中没有解析到有效数据")
            
            df = pd.DataFrame(data_list)
            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线")
            if len(df_backtest) > 0:
                print(f"   数据区间: {df_backtest.index[0]} 至 {df_backtest.index[-1]}")
            
            return df_backtest
            
        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()