#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 只做多T+1自动化交易报告系统
基于 cyb50_30min_long_only_t1.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
)
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 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"""
🚀 创业板50交易报告 (T+1)
生成时间: {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
# 计算统计数据 - 使用实际盈亏总和计算最终资金（更准确）
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_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"""
🚀 创业板50交易报告 (T+1)
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
数据区间: 近2个月
数据来源: 多空双向策略的做多交易 + T+1规则转换
📊 总体绩效
| 指标 | 数值 |
|---|
| 初始资金 | {initial_capital:,.0f}元 |
| 最终资金 | {final_capital:,.0f}元 |
| 总收益率 | {total_return:+.2f}% |
| 总交易次数 | {total_trades}笔 |
| 胜率 | {win_rate:.1f}% |
| 盈亏比 | {profit_factor:.2f} |
🔄 交易统计 (T+1调整影响)
| 类型 | 交易次数 | 胜率 | 总盈亏 |
|---|
| 正常交易 | {normal_count}笔 |
{(normal_win/normal_count*100 if normal_count>0 else 0):.1f}% |
{normal_pnl:+,.0f}元 |
| T+1调整(T0→T1) | {t1_count}笔 |
{(t1_win/t1_count*100 if t1_count>0 else 0):.1f}% |
{t1_pnl:+,.0f}元 |
📈 T+1调整明细
| 开仓时间 | 原平仓 | 新平仓 | 原盈亏 | 新盈亏 | 变化 |
|---|
"""
for _, trade in t1_adjusted.iterrows():
change = trade['盈亏变化']
change_class = "positive" if change >= 0 else "negative"
html += f"""
| {trade['开仓时间'].strftime('%m-%d %H:%M')} |
{trade['原平仓时间'].strftime('%m-%d %H:%M')} |
{trade['平仓时间'].strftime('%m-%d %H:%M')} |
{trade['原盈亏']:+.0f} |
{trade['盈亏金额']:+.0f} |
{change:+.0f} |
"""
if len(t1_adjusted) == 0:
html += "| 无T+1调整交易 |
"
html += """
📝 最近30笔交易明细
| 开仓时间 | 平仓时间 | 开仓价 | 平仓价 |
盈亏 | 退出原因 | T+1调整 |
|---|
"""
for _, trade in trades_df.tail(30).iterrows():
pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative"
t1_flag = "✓" if trade['T+1调整'] == '是(T0→T1)' else ""
html += f"""
| {trade['开仓时间'].strftime('%m-%d %H:%M')} |
{trade['平仓时间'].strftime('%m-%d %H:%M')} |
{trade['开仓价格']:.2f} |
{trade['平仓价格']:.2f} |
{trade['盈亏金额']:+.0f} |
{trade['退出原因']} |
{t1_flag} |
"""
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: 获取近3个月数据...")
try:
config_manager = ConfigManager('config.json')
fetcher = IntradayDataFetcher(config_manager)
end_date = datetime.now()
start_date = end_date - timedelta(days=90) # 近3个月
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_v2(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()