#!/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"""

🚀 创业板50交易报告

生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

数据区间: 近2个月

近2个月无交易信号触发

初始资金: {initial_capital:,.0f}元

最终资金: {final_capital:,.0f}元

收益率: {total_return:+.2f}%

""" 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"""

🚀 创业板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}

🔄 多空统计

方向交易次数胜率总盈亏
做多{len(long_trades)}笔 {(len(long_trades[long_trades['盈亏金额']>0])/len(long_trades)*100 if len(long_trades)>0 else 0):.1f}% {long_trades['盈亏金额'].sum():+,.0f}元
做空{len(short_trades)}笔 {(len(short_trades[short_trades['盈亏金额']>0])/len(short_trades)*100 if len(short_trades)>0 else 0):.1f}% {short_trades['盈亏金额'].sum():+,.0f}元

📝 最近30笔交易明细

""" for _, trade in trades_df.tail(30).iterrows(): pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative" html += f""" """ html += "
方向开仓时间平仓时间开仓价平仓价 盈亏退出原因
{trade['交易方向']} {trade['开仓时间']} {trade['平仓时间']} {trade['开仓价格']:.2f} {trade['平仓价格']:.2f} {trade['盈亏金额']:+.0f} {trade['退出原因']}
" # 纯文本报告 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()