cyb50-quant/cat-fly/auto_report_realtime.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 只做多T+1自动化交易报告系统 (实时信号增强版)
在14:50等时间点会使用实时行情数据计算当前信号
"""

import sys
sys.path.insert(0, '/root/.openclaw/workspace/cat-fly')

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

# 导入策略模块
from cyb50_30min_dual_direction import (
    ConfigManager, IntradayDataFetcher, 
    DualDirectionSignalGenerator, DualDirectionExecutor
)
from t1_converter import simulate_t1_trades_v2, compare_results

# ==================== 邮件配置 ====================
EMAIL_CONFIG = {
    "smtp_server": "localhost",
    "smtp_port": 25,
    "sender_email": "cyb50-t1@erwin.wang",
    "receiver_emails": ["380880504@qq.com", "1095512042@qq.com"]
}

def send_email(subject, html_content, text_content=""):
    """发送邮件"""
    try:
        msg = MIMEMultipart('alternative')
        msg['Subject'] = Header(subject, 'utf-8')
        msg['From'] = EMAIL_CONFIG['sender_email']
        msg['To'] = ', '.join(EMAIL_CONFIG['receiver_emails'])
        
        text_part = MIMEText(text_content, 'plain', 'utf-8')
        msg.attach(text_part)
        
        html_part = MIMEText(html_content, 'html', 'utf-8')
        msg.attach(html_part)
        
        with smtplib.SMTP(EMAIL_CONFIG['smtp_server'], EMAIL_CONFIG['smtp_port']) as server:
            server.sendmail(
                EMAIL_CONFIG['sender_email'],
                EMAIL_CONFIG['receiver_emails'],
                msg.as_string()
            )
        print(f"✅ 邮件发送成功: {subject}")
        return True
    except Exception as e:
        print(f"❌ 邮件发送失败: {e}")
        return False


def get_realtime_price():
    """获取创业板50实时价格"""
    try:
        import akshare as ak
        # 使用新浪接口获取实时行情（更稳定）
        df = ak.stock_zh_index_spot_sina()
        cyb50 = df[df['代码'] == 'sz399673']
        if len(cyb50) > 0:
            return {
                'price': float(cyb50.iloc[0]['最新价']),
                'open': float(cyb50.iloc[0]['今开']),
                'high': float(cyb50.iloc[0]['最高']),
                'low': float(cyb50.iloc[0]['最低']),
                'volume': float(cyb50.iloc[0]['成交量']),
                'change_pct': float(cyb50.iloc[0].get('涨跌幅', 0)),
                'time': datetime.now().strftime('%H:%M:%S')
            }
    except Exception as e:
        print(f"获取实时行情失败: {e}")
    return None


def calculate_realtime_signal(current_price, last_kline_data):
    """
    基于最新价格和最后一根K线数据，估算当前信号状态
    """
    score = 0
    signals = []
    
    last_close = last_kline_data['Close']
    price_change_pct = (current_price - last_close) / last_close
    
    # 1. RSI估算
    estimated_rsi = last_kline_data['RSI'] + price_change_pct * 250
    if estimated_rsi < 30:
        score += 2
        signals.append(f"RSI超卖(估{estimated_rsi:.1f})")
    elif estimated_rsi < 35:
        score += 1
        signals.append(f"RSI偏弱(估{estimated_rsi:.1f})")
    
    # 2. KDJ估算
    estimated_j = last_kline_data['J'] + price_change_pct * 300
    if estimated_j < 0:
        score += 1
        signals.append(f"KDJ极端超卖(估J={estimated_j:.1f})")
    
    # 3. 布林带位置
    bb_lower = last_kline_data['BB_lower']
    if current_price <= bb_lower * 1.005:
        score += 2
        signals.append("触及下轨")
    elif current_price <= bb_lower * 1.02:
        score += 1
        signals.append("接近下轨")
    
    # 4. 价格动量
    if price_change_pct < -0.015:
        score += 1
        signals.append(f"动量超卖({price_change_pct*100:.2f}%)")
    
    # 5. MA趋势
    if last_kline_data['MA6'] > last_kline_data['MA12']:
        score += 1
        signals.append("MA短期上行")
    else:
        score -= 1
        signals.append("MA下降趋势惩罚")
    
    return {
        'score': score,
        'signals': signals,
        'estimated_rsi': estimated_rsi,
        'estimated_j': estimated_j,
        'price_change_pct': price_change_pct,
        'bb_lower': bb_lower,
        'bb_upper': last_kline_data['BB_upper'],
        'current_price': current_price,
        'last_close': last_close,
        'triggered': score >= 4
    }


def is_pre_close_time():
    """检查是否是收盘前10分钟(14:50左右)"""
    now = datetime.now()
    return now.hour == 14 and now.minute >= 50


def check_today_trades(trades_df):
    """检查当天是否有交易"""
    if len(trades_df) == 0:
        return False, pd.DataFrame()
    
    today = datetime.now().date()
    today_trades = trades_df[
        (pd.to_datetime(trades_df['开仓时间']).dt.date == today) | 
        (pd.to_datetime(trades_df['平仓时间']).dt.date == today)
    ]
    
    has_today_trade = len(today_trades) > 0
    
    if has_today_trade:
        print(f"📊 当天交易数量: {len(today_trades)}笔")
        for _, trade in today_trades.iterrows():
            print(f"   {trade['开仓时间']} → {trade['平仓时间']} | {trade['盈亏金额']:+.0f}元")
    else:
        print("📭 当天无交易")
    
    return has_today_trade, today_trades


def is_post_close_time():
    """检查是否是盘后时间（15:00-15:30）"""
    now = datetime.now()
    return now.hour == 15 and now.minute >= 0 and now.minute <= 30


def generate_report(trades_df, initial_capital=1000000, realtime_signal=None):
    """生成只做多T+1交易报告（增强版，包含实时信号）"""
    
    if len(trades_df) == 0:
        final_capital = initial_capital
        total_return = 0
    else:
        total_pnl = trades_df['盈亏金额'].sum()
        final_capital = initial_capital + total_pnl
        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
    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
    
    # T+1调整统计
    t1_adjusted = trades_df[trades_df['T+1调整'] == '是(T0→T1)']
    t1_count = len(t1_adjusted)
    t1_pnl = t1_adjusted['盈亏金额'].sum() if len(t1_adjusted) > 0 else 0
    
    # 构建HTML报告
    now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # 实时信号部分
    realtime_html = ""
    if realtime_signal:
        signal_color = "green" if realtime_signal['triggered'] else "orange" if realtime_signal['score'] >= 3 else "gray"
        signal_text = "🟢 触发买入" if realtime_signal['triggered'] else "🟡 接近触发" if realtime_signal['score'] >= 3 else "⚪ 未触发"
        
        realtime_html = f"""
        <h2>🔔 实时信号检测 (收盘前)</h2>
        <table>
            <tr><th>指标</th><th>数值</th></tr>
            <tr><td>检测时间</td><td>{now_str}</td></tr>
            <tr><td>实时价格</td><td>{realtime_signal['current_price']:.2f}</td></tr>
            <tr><td>较上一K线</td><td>{realtime_signal['price_change_pct']*100:+.2f}%</td></tr>
            <tr><td>RSI(估算)</td><td>{realtime_signal['estimated_rsi']:.2f}</td></tr>
            <tr><td>KDJ J(估算)</td><td>{realtime_signal['estimated_j']:.2f}</td></tr>
            <tr><td>布林带下轨</td><td>{realtime_signal['bb_lower']:.2f}</td></tr>
            <tr><td>信号评分</td><td style="color: {signal_color}; font-weight: bold;">{realtime_signal['score']}/4</td></tr>
            <tr><td>触发信号</td><td>{', '.join(realtime_signal['signals']) if realtime_signal['signals'] else '无'}</td></tr>
            <tr><td>最终判断</td><td style="color: {signal_color}; font-size: 16px; font-weight: bold;">{signal_text}</td></tr>
        </table>
        """
    
    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; }}
    .highlight {{ background-color: #fff3cd; }}
    </style></head><body>
    <h1>🚀 创业板50交易报告 (T+1)</h1>
    <p>生成时间: {now_str}</p>
    <p>数据区间: 近3个月</p>
    {realtime_html}
    
    <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>
    </table>
    
    <h2>📝 最近交易明细</h2>
    <table>
        <tr><th>开仓时间</th><th>平仓时间</th><th>开仓价</th><th>平仓价</th>
            <th>盈亏</th><th>退出原因</th><th>T+1调整</th></tr>
    """
    
    for _, trade in trades_df.tail(10).iterrows():
        pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative"
        t1_flag = "✓" if trade['T+1调整'] == '是(T0→T1)' else ""
        html += f"""
        <tr>
            <td>{trade['开仓时间'].strftime('%m-%d %H:%M')}</td>
            <td>{trade['平仓时间'].strftime('%m-%d %H:%M')}</td>
            <td>{trade['开仓价格']:.2f}</td>
            <td>{trade['平仓价格']:.2f}</td>
            <td class="{pnl_class}">{trade['盈亏金额']:+.0f}</td>
            <td>{trade['退出原因']}</td>
            <td>{t1_flag}</td>
        </tr>
        """
    
    if len(trades_df) == 0:
        html += "<tr><td colspan='7'>近2个月无交易信号触发</td></tr>"
    
    html += "</table></body></html>"
    
    # 纯文本报告
    text = f"""
创业板50交易报告 (T+1)
生成时间: {now_str}

{f"【实时信号检测】\n时间: {now_str}\n实时价格: {realtime_signal['current_price']:.2f}\n信号评分: {realtime_signal['score']}/4\n触发信号: {', '.join(realtime_signal['signals']) if realtime_signal['signals'] else '无'}\n判断: {'触发买入' if realtime_signal['triggered'] else '未触发'}\n" if realtime_signal else ""}
【总体绩效】
初始资金: {initial_capital:,.0f}元
最终资金: {final_capital:,.0f}元
总收益率: {total_return:+.2f}%
总交易次数: {total_trades}笔
胜率: {win_rate:.1f}%
"""
    
    return html, text, final_capital


def main():
    """主程序 - 实时信号增强版"""
    print("="*80)
    print("🚀 创业板50只做多T+1自动交易报告系统 (实时信号增强版)")
    print("="*80)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    initial_capital = 1000000
    realtime_signal = None
    
    # 判断是否是收盘前时间
    pre_close = is_pre_close_time()
    if pre_close:
        print("\n🔔 检测到收盘前时间(14:50+)，将获取实时行情计算信号...")
    
    # 1. 获取数据
    print("\n📊 步骤1: 获取近3个月数据...")
    try:
        config_manager = ConfigManager('config.json')
        fetcher = IntradayDataFetcher(config_manager)
        end_date = datetime.now()
        start_date = end_date - timedelta(days=90)
        raw_data = fetcher.fetch_30min_data(start_date, end_date)
        
        if raw_data is None or len(raw_data) == 0:
            print("❌ 数据获取失败")
            return
        
        print(f"✅ 数据获取成功: {len(raw_data)}条K线")
        print(f"   数据区间: {raw_data.index[0]} ~ {raw_data.index[-1]}")
        
    except Exception as e:
        print(f"❌ 数据获取失败: {e}")
        import traceback
        traceback.print_exc()
        return
    
    # 2. 如果是收盘前时间，获取实时信号
    if pre_close:
        print("\n📈 步骤1.5: 获取实时行情并计算信号...")
        try:
            realtime = get_realtime_price()
            if realtime:
                print(f"   实时价格: {realtime['price']:.2f} (时间: {realtime['time']})")
                
                # 计算技术指标
                data_with_indicators = fetcher.calculate_intraday_indicators(raw_data)
                last_kline = data_with_indicators.iloc[-1]
                
                # 计算实时信号
                realtime_signal = calculate_realtime_signal(realtime['price'], last_kline)
                print(f"   实时信号评分: {realtime_signal['score']}/4")
                print(f"   触发信号: {', '.join(realtime_signal['signals']) if realtime_signal['signals'] else '无'}")
                print(f"   是否触发: {'是' if realtime_signal['triggered'] else '否'}")
            else:
                print("   ⚠️ 获取实时行情失败，跳过实时信号计算")
        except Exception as e:
            print(f"   ⚠️ 实时信号计算失败: {e}")
    
    # 3. 生成多空双向信号并执行
    print("\n📈 步骤2: 运行多空双向策略获取做多交易...")
    try:
        data_with_indicators = fetcher.calculate_intraday_indicators(raw_data)
        signal_generator = DualDirectionSignalGenerator()
        signals_df = signal_generator.generate_dual_direction_signals(data_with_indicators)
        
        executor = DualDirectionExecutor(initial_capital=initial_capital)
        results_df, trades_df = executor.execute_dual_direction_trades(signals_df)
        
        long_trades = trades_df[trades_df['交易方向'] == '做多'].copy()
        print(f"✅ 多空策略完成: 共{len(trades_df)}笔交易, 做多{len(long_trades)}笔")
        
    except Exception as e:
        print(f"❌ 多空策略执行失败: {e}")
        import traceback
        traceback.print_exc()
        return
    
    # 4. 应用T+1规则转换
    print("\n🔄 步骤3: 应用T+1规则转换...")
    try:
        t1_trades = simulate_t1_trades_v2(data_with_indicators, long_trades, initial_capital)
        print(f"✅ T+1转换完成: {len(t1_trades)}笔交易")
    except Exception as e:
        print(f"❌ T+1转换失败: {e}")
        import traceback
        traceback.print_exc()
        return
    
    # 5. 检查当天交易并决定是否发送邮件
    print("\n📊 步骤4: 检查当天交易情况...")
    has_today_trade, today_trades = check_today_trades(t1_trades)
    
    # 判断是否应该发送邮件
    should_send = False
    send_reason = ""
    
    if has_today_trade:
        should_send = True
        send_reason = f"当天有{len(today_trades)}笔交易"
    elif realtime_signal and realtime_signal['triggered']:
        should_send = True
        send_reason = "实时信号触发买入"
    elif pre_close:
        should_send = True
        send_reason = "收盘前例行报告（含实时信号）"
    elif is_post_close_time():
        should_send = True
        send_reason = "盘后时间，当天无交易，发送例行报告"
    else:
        should_send = False
        send_reason = "当天无交易，非关键时间点，跳过发送"
    
    print(f"\n📧 邮件发送决策: {send_reason}")
    
    if should_send:
        # 6. 生成报告（含实时信号）
        print("\n📝 步骤5: 生成报告...")
        html_report, text_report, final_capital = generate_report(t1_trades, initial_capital, realtime_signal)
        
        # 7. 发送邮件
        print("\n📧 步骤6: 发送邮件...")
        total_trades = len(t1_trades)
        total_return = (final_capital/initial_capital-1)*100
        
        # 邮件主题包含实时信号信息
        if realtime_signal:
            sig_emoji = "🟢" if realtime_signal['triggered'] else "⚪"
            subject = f"{sig_emoji} CYB50-T1报告 {datetime.now().strftime('%m-%d %H:%M')} | 实时{realtime_signal['score']}分 | 收益{total_return:+.2f}%"
        else:
            subject = f"🚀 CYB50-T1报告 {datetime.now().strftime('%m-%d %H:%M')} | 收益{total_return:+.2f}% | {total_trades}笔交易"
        
        send_email(subject, html_report, text_report)
    else:
        print("⏭️  跳过邮件发送")
    
    print("\n✅ 全部完成!")
    print("="*80)


if __name__ == "__main__":
    main()