openclaw
/
cyb50-quant


			
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332
							#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 自动化交易报告系统
基于 cyb50_30min_dual_direction.py 完整策略
"""

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
)

# ==================== 邮件配置 ====================
EMAIL_CONFIG = {
    "smtp_server": "localhost",
    "smtp_port": 25,
    "sender_email": "catfly@erwin.wang",
    "receiver_emails": ["380880504@qq.com", "695047456@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 check_today_trades(trades_df):
    """检查当天是否有交易
    
    返回:
        has_today_trade: bool, 当天是否有交易
        today_trades: DataFrame, 当天的交易记录
    """
    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['平仓时间']} | {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, results_df, initial_capital=1000000):
    """生成交易报告"""
    
    if len(trades_df) == 0:
        final_capital = initial_capital
        total_return = 0
        html = f"""
        <html><body>
        <h1>🚀 创业板50交易报告</h1>
        <p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
        <p>数据区间: 近2个月</p>
        <p><b>近2个月无交易信号触发</b></p>
        <p>初始资金: {initial_capital:,.0f}元</p>
        <p>最终资金: {final_capital:,.0f}元</p>
        <p>收益率: {total_return:+.2f}%</p>
        </body></html>
        """
        text = f"近2个月无交易信号\n初始资金: {initial_capital:,.0f}元\n最终资金: {final_capital:,.0f}元"
        return html, text, final_capital
    
    # 计算统计数据
    final_capital = results_df['net_value'].iloc[-1]
    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
    
    long_trades = trades_df[trades_df['交易方向'] == '做多']
    short_trades = trades_df[trades_df['交易方向'] == '做空']
    
    # 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; }}
    </style></head><body>
    <h1>🚀 创业板50交易报告</h1>
    <p>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
    <p>数据区间: 近2个月</p>
    
    <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></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>📝 最近30笔交易明细</h2>
    <table>
        <tr><th>方向</th><th>开仓时间</th><th>平仓时间</th><th>开仓价</th><th>平仓价</th>
            <th>盈亏</th><th>退出原因</th></tr>
    """
    
    for _, trade in trades_df.tail(30).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>{trade['退出原因']}</td>
        </tr>
        """
    
    html += "</table></body></html>"
    
    # 纯文本报告
    text = f"""
创业板50交易报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

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

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

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


def main():
    """主程序"""
    print("="*80)
    print("🚀 cat-fly 自动交易报告系统 (完整版)")
    print("="*80)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    initial_capital = 1000000
    
    # 1. 获取数据
    print("\n📊 步骤1: 获取近2个月数据...")
    try:
        # 使用原版数据获取器
        fetcher = IntradayDataFetcher()
        end_date = datetime.now()
        start_date = end_date - timedelta(days=70)  # 2个月+10天缓冲
        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}")
        return
    
    # 2. 生成信号
    print("\n📈 步骤2: 生成交易信号...")
    try:
        # 先计算技术指标（使用数据获取器的方法）
        print("   计算技术指标...")
        data_with_indicators = fetcher.calculate_intraday_indicators(raw_data)
        
        # 生成信号
        signal_generator = DualDirectionSignalGenerator()
        signals_df = signal_generator.generate_dual_direction_signals(data_with_indicators)
        
        # 统计信号
        long_signals = len(signals_df[signals_df['Signal'] == 1])
        short_signals = len(signals_df[signals_df['Signal'] == -1])
        print(f"✅ 信号生成完成: 做多信号{long_signals}个, 做空信号{short_signals}个")
        
    except Exception as e:
        print(f"❌ 信号生成失败: {e}")
        return
    
    # 3. 执行回测
    print("\n💼 步骤3: 执行交易回测...")
    try:
        executor = DualDirectionExecutor(initial_capital=initial_capital)
        results_df, trades_df = executor.execute_dual_direction_trades(signals_df)
        
        print(f"✅ 回测完成: 共执行{len(trades_df)}笔交易")
        if len(trades_df) > 0:
            final_value = results_df['net_value'].iloc[-1]
            print(f"   最终资金: {final_value:,.0f}元")
            print(f"   收益率: {(final_value/initial_capital-1)*100:+.2f}%")
        
    except Exception as e:
        print(f"❌ 回测执行失败: {e}")
        return
    
    # 4. 检查当天交易并决定是否发送邮件
    print("\n📊 步骤4: 检查当天交易情况...")
    has_today_trade, today_trades = check_today_trades(trades_df)
    
    # 判断是否应该发送邮件
    should_send = False
    send_reason = ""
    
    if has_today_trade:
        # 有当天交易，正常发送
        should_send = True
        send_reason = f"当天有{len(today_trades)}笔交易"
    elif is_post_close_time():
        # 没有当天交易，但在盘后时间（15:00-15:30），发送一次
        should_send = True
        send_reason = "盘后时间，当天无交易，发送例行报告"
    else:
        # 没有当天交易，也不在盘后时间，跳过发送
        should_send = False
        send_reason = "当天无交易，非盘后时间，跳过发送"
    
    print(f"\n📧 邮件发送决策: {send_reason}")
    
    if should_send:
        # 5. 生成报告
        print("\n📝 步骤5: 生成报告...")
        html_report, text_report, final_capital = generate_report(trades_df, results_df, initial_capital)
        
        # 6. 发送邮件
        print("\n📧 步骤6: 发送邮件...")
        total_trades = len(trades_df)
        total_return = (final_capital/initial_capital-1)*100
        subject = f"🚀 创业板50报告 {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()