openclaw
/
cyb50-quant


			
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398
							#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 只做多T+1自动化交易报告系统
基于 cyb50_30min_long_only_t1.py 策略
"""

import sys
import os

# 设置 stdout 编码为 utf-8
if sys.platform == 'win32':
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

# 调试版本：所有依赖文件在同一目录下，无需额外路径设置
# 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, compare_results

# ==================== 邮件配置 ====================
EMAIL_CONFIG = {
    "smtp_server": "localhost",
    "smtp_port": 25,
    "sender_email": "cyb50-t1@erwin.wang",
    "receiver_emails": ["380880504@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['盈亏金额']:+.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):
    """生成只做多T+1交易报告（基于多空版本转换后的数据）"""
    
    if len(trades_df) == 0:
        final_capital = initial_capital
        total_return = 0
        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交易报告 (T+1)</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 = trades_df['平仓时资金'].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
    
    # T+1调整统计
    t1_adjusted = trades_df[trades_df['T+1调整'] == '是(T0→T1)']
    t1_count = len(t1_adjusted)
    t1_win = len(t1_adjusted[t1_adjusted['盈亏金额'] > 0]) if len(t1_adjusted) > 0 else 0
    t1_pnl = t1_adjusted['盈亏金额'].sum() if len(t1_adjusted) > 0 else 0
    
    # 非T+1调整交易
    t1_normal = trades_df[trades_df['T+1调整'] != '是(T0→T1)']
    normal_count = len(t1_normal)
    normal_win = len(t1_normal[t1_normal['盈亏金额'] > 0]) if len(t1_normal) > 0 else 0
    normal_pnl = t1_normal['盈亏金额'].sum() if len(t1_normal) > 0 else 0
    
    # HTML报告 - 使用原版cat-fly蓝色主题
    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>生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
    <p>数据区间: 近2个月</p>
    <p><b>数据来源: 多空双向策略的做多交易 + T+1规则转换</b></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>🔄 交易统计 (T+1调整影响)</h2>
    <table>
        <tr><th>类型</th><th>交易次数</th><th>胜率</th><th>总盈亏</th></tr>
        <tr>
            <td>正常交易</td><td>{normal_count}笔</td>
            <td>{(normal_win/normal_count*100 if normal_count>0 else 0):.1f}%</td>
            <td class="{'positive' if normal_pnl >= 0 else 'negative'}">{normal_pnl:+,.0f}元</td>
        </tr>
        <tr class="{'highlight' if t1_count > 0 else ''}">
            <td>T+1调整(T0→T1)</td><td>{t1_count}笔</td>
            <td>{(t1_win/t1_count*100 if t1_count>0 else 0):.1f}%</td>
            <td class="{'positive' if t1_pnl >= 0 else 'negative'}">{t1_pnl:+,.0f}元</td>
        </tr>
    </table>
    
    <h2>📈 T+1调整明细</h2>
    <table>
        <tr><th>开仓时间</th><th>原平仓</th><th>新平仓</th><th>原盈亏</th><th>新盈亏</th><th>变化</th></tr>
    """
    
    for _, trade in t1_adjusted.iterrows():
        change = trade['盈亏变化']
        change_class = "positive" if change >= 0 else "negative"
        html += f"""
        <tr>
            <td>{trade['开仓时间'].strftime('%m-%d %H:%M')}</td>
            <td>{trade['原平仓时间'].strftime('%m-%d %H:%M')}</td>
            <td>{trade['平仓时间'].strftime('%m-%d %H:%M')}</td>
            <td>{trade['原盈亏']:+.0f}</td>
            <td class="{'positive' if trade['盈亏金额'] >= 0 else 'negative'}">{trade['盈亏金额']:+.0f}</td>
            <td class="{change_class}">{change:+.0f}</td>
        </tr>
        """
    
    if len(t1_adjusted) == 0:
        html += "<tr><td colspan='6'>无T+1调整交易</td></tr>"
    
    html += """
    </table>
    
    <h2>📝 最近20笔交易明细</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(20).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>
        """
    
    html += "</table></body></html>"
    
    # 纯文本报告
    text = f"""
创业板50交易报告 (T+1)
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
数据来源: 多空双向策略的做多交易 + T+1规则转换

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

【T+1调整统计】
正常交易: {normal_count}笔, 盈亏{normal_pnl:+,.0f}元
T+1调整(T0→T1): {t1_count}笔, 盈亏{t1_pnl:+,.0f}元

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


def main():
    """主程序 - 使用多空版本做多交易 + T+1规则转换"""
    print("="*80)
    print("🚀 创业板50只做多T+1自动交易报告系统")
    print("(基于多空双向策略的做多交易 + T+1规则转换)")
    print("="*80)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    initial_capital = 1000000
    
    # 1. 获取数据
    print("\n📊 步骤1: 获取近2个月数据...")
    try:
        config_manager = ConfigManager('config.json')
        fetcher = IntradayDataFetcher(config_manager)
        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}")
        import traceback
        traceback.print_exc()
        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)
        
        # 执行多空双向交易
        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
    
    # 3. 应用T+1规则转换
    print("\n🔄 步骤3: 应用T+1规则转换...")
    try:
        t1_trades = simulate_t1_trades(data_with_indicators, long_trades, initial_capital)
        print(f"✅ T+1转换完成: {len(t1_trades)}笔交易")
        
        # 统计T+1调整
        t1_adjusted = t1_trades[t1_trades['T+1调整'] == '是(T0→T1)']
        print(f"   T+1调整交易: {len(t1_adjusted)}笔")
        
    except Exception as e:
        print(f"❌ T+1转换失败: {e}")
        import traceback
        traceback.print_exc()
        return
    
    # 4. 检查当天交易并决定是否发送邮件
    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 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(t1_trades, initial_capital)
        
        # 6. 发送邮件
        print("\n📧 步骤6: 发送邮件...")
        total_trades = len(t1_trades)
        total_return = (final_capital/initial_capital-1)*100
        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()