#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 创业板50指数 - 自动化交易报告系统 (独立版) 功能： 1. 获取近2个月数据 2. 运行策略回测 3. 生成详细报告 4. 发送邮件通知 执行频率：A股开盘时间每半小时（9:30-11:30, 13:00-15:00） """ import pandas as pd import numpy as np import akshare as ak import warnings import os import smtplib import ssl from datetime import datetime, timedelta from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header warnings.filterwarnings('ignore') # ==================== 邮件配置 ==================== # 请修改以下配置为你的邮箱信息 EMAIL_CONFIG = { "smtp_server": "smtp.qq.com", # SMTP服务器 "smtp_port": 465, # SMTP端口 "sender_email": "your_email@qq.com", # 发件人邮箱 "sender_password": "your_auth_code", # 邮箱授权码（不是登录密码） "receiver_email": "your_email@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'] = EMAIL_CONFIG['receiver_email'] # 纯文本版本 text_part = MIMEText(text_content, 'plain', 'utf-8') msg.attach(text_part) # HTML版本 html_part = MIMEText(html_content, 'html', 'utf-8') msg.attach(html_part) # 发送邮件 context = ssl.create_default_context() with smtplib.SMTP_SSL(EMAIL_CONFIG['smtp_server'], EMAIL_CONFIG['smtp_port'], context=context) as server: server.login(EMAIL_CONFIG['sender_email'], EMAIL_CONFIG['sender_password']) server.sendmail( EMAIL_CONFIG['sender_email'], EMAIL_CONFIG['receiver_email'], msg.as_string() ) print(f"✅ 邮件发送成功: {subject}") return True except Exception as e: print(f"❌ 邮件发送失败: {e}") print(f" 请检查EMAIL_CONFIG配置是否正确") return False # ==================== 数据获取 ==================== class DataFetcher: """数据获取类""" @staticmethod def fetch_recent_2months(): """获取近2个月数据""" end_date = datetime.now() start_date = end_date - timedelta(days=70) # 2个月+10天缓冲 print(f"获取数据: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}") try: # 使用akshare获取30分钟K线 df = ak.index_zh_a_hist_min_em( symbol="399673", period="30", start_date=start_date.strftime('%Y%m%d%H%M%S'), end_date=end_date.strftime('%Y%m%d%H%M%S') ) if df is None or len(df) == 0: print("❌ 未获取到数据") return None # 标准化列名 df = df.rename(columns={ '时间': 'datetime', '开盘': 'open', '收盘': 'close', '最高': 'high', '最低': 'low', '成交量': 'volume' }) df['datetime'] = pd.to_datetime(df['datetime']) df = df.set_index('datetime').sort_index() # 只保留最近2个月的数据用于回测 backtest_start = end_date - timedelta(days=60) df_backtest = df[df.index >= backtest_start] print(f"✅ 数据获取成功: 共{len(df_backtest)}条30分钟K线") print(f" 数据区间: {df_backtest.index[0]} 至 {df_backtest.index[-1]}") return df_backtest except Exception as e: print(f"❌ 数据获取失败: {e}") return None # ==================== 策略类 ==================== class CatFlyStrategy: """cat-fly策略简化版 - 基于30分钟K线""" def __init__(self, config=None): self.config = config or { 'initial_capital': 1000000, 'position_size_pct': 1.0, 'stop_loss_pct': 0.008, 'take_profit_pct': 0.02, 'max_hold_bars': 16, 'min_signal_strength': 3 } self.initial_capital = self.config['initial_capital'] def calculate_indicators(self, df): """计算技术指标""" # RSI delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['RSI'] = 100 - (100 / (1 + rs)) # 移动平均线 df['MA5'] = df['close'].rolling(5).mean() df['MA20'] = df['close'].rolling(20).mean() df['MA60'] = df['close'].rolling(60).mean() # 布林带 df['BB_middle'] = df['close'].rolling(20).mean() bb_std = df['close'].rolling(20).std() df['BB_upper'] = df['BB_middle'] + 2 * bb_std df['BB_lower'] = df['BB_middle'] - 2 * bb_std # MACD ema12 = df['close'].ewm(span=12).mean() ema26 = df['close'].ewm(span=26).mean() df['MACD'] = ema12 - ema26 df['MACD_signal'] = df['MACD'].ewm(span=9).mean() return df def generate_signals(self, df): """生成交易信号""" df = self.calculate_indicators(df) df['signal'] = 0 df['signal_strength'] = 0 for i in range(60, len(df)): row = df.iloc[i] strength = 0 # RSI超卖/超买 if row['RSI'] < 30: strength += 1 elif row['RSI'] > 70: strength -= 1 # 均线多头排列/空头排列 if row['close'] > row['MA5'] > row['MA20']: strength += 1 elif row['close'] < row['MA5'] < row['MA20']: strength -= 1 # 布林带 if row['close'] < row['BB_lower']: strength += 1 elif row['close'] > row['BB_upper']: strength -= 1 # MACD金叉/死叉 if i > 0: prev_macd = df['MACD'].iloc[i-1] prev_signal = df['MACD_signal'].iloc[i-1] curr_macd = row['MACD'] curr_signal_line = row['MACD_signal'] if prev_macd < prev_signal and curr_macd > curr_signal_line: strength += 1 elif prev_macd > prev_signal and curr_macd < curr_signal_line: strength -= 1 df.iloc[i, df.columns.get_loc('signal_strength')] = strength # 生成交易信号 if strength >= self.config['min_signal_strength']: df.iloc[i, df.columns.get_loc('signal')] = 1 # 做多 elif strength <= -self.config['min_signal_strength']: df.iloc[i, df.columns.get_loc('signal')] = -1 # 做空 return df def backtest(self, df): """回测""" df = self.generate_signals(df) trades = [] capital = self.initial_capital position = 0 entry_price = 0 entry_time = None holding_bars = 0 for i in range(60, len(df)): current_bar = df.iloc[i] price = current_bar['close'] current_time = current_bar.name # 无持仓时检查开仓信号 if position == 0: if current_bar['signal'] == 1: # 做多 position_size = int(capital * self.config['position_size_pct'] / price) if position_size > 0: position = position_size entry_price = price entry_time = current_time holding_bars = 0 elif current_bar['signal'] == -1: # 做空 position_size = int(capital * self.config['position_size_pct'] / price) if position_size > 0: position = -position_size entry_price = price entry_time = current_time holding_bars = 0 # 有持仓时检查平仓 else: holding_bars += 1 exit_signal = False exit_reason = "" if position > 0: # 做多持仓 if price <= entry_price * (1 - self.config['stop_loss_pct']): exit_signal = True exit_reason = "止损" elif price >= entry_price * (1 + self.config['take_profit_pct']): exit_signal = True exit_reason = "止盈" elif holding_bars >= self.config['max_hold_bars']: exit_signal = True exit_reason = "时间止损" elif current_bar['RSI'] > 70: exit_signal = True exit_reason = "信号消失(RSI超买)" else: # 做空持仓 if price >= entry_price * (1 + self.config['stop_loss_pct']): exit_signal = True exit_reason = "止损" elif price <= entry_price * (1 - self.config['take_profit_pct']): exit_signal = True exit_reason = "止盈" elif holding_bars >= self.config['max_hold_bars']: exit_signal = True exit_reason = "时间止损" elif current_bar['RSI'] < 30: exit_signal = True exit_reason = "信号消失(RSI超卖)" # 执行平仓 if exit_signal: if position > 0: pnl = (price - entry_price) * position pnl_pct = (price - entry_price) / entry_price * 100 else: pnl = (entry_price - price) * abs(position) pnl_pct = (entry_price - price) / entry_price * 100 capital += pnl trades.append({ '方向': '做多' if position > 0 else '做空', '开仓时间': entry_time, '平仓时间': current_time, '开仓价': entry_price, '平仓价': price, '持仓数量': abs(position), '盈亏金额': pnl, '盈亏百分比': pnl_pct, '退出原因': exit_reason, '持仓周期': holding_bars, '平仓后资金': capital }) position = 0 entry_price = 0 entry_time = None holding_bars = 0 return df, pd.DataFrame(trades), capital # ==================== 报告生成 ==================== def generate_report(trades_df, final_capital, initial_capital=1000000): """生成详细报告""" if len(trades_df) == 0: html = "

创业板50交易报告

近2个月无交易信号

" text = "近2个月无交易信号" return html, text 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 avg_profit = winning_trades['盈亏金额'].mean() if len(winning_trades) > 0 else 0 avg_loss = losing_trades['盈亏金额'].mean() if len(losing_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 max_profit = trades_df['盈亏金额'].max() max_loss = trades_df['盈亏金额'].min() avg_hold_time = trades_df['持仓周期'].mean() long_trades = trades_df[trades_df['方向'] == '做多'] short_trades = trades_df[trades_df['方向'] == '做空'] exit_reasons = trades_df['退出原因'].value_counts() # 生成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}
平均持仓时间{avg_hold_time:.1f}周期 ({avg_hold_time*0.5:.1f}小时)

📈 盈亏统计

指标数值
总盈利+{total_profit:,.0f}元
总亏损-{total_loss:,.0f}元
平均盈利+{avg_profit:,.0f}元
平均亏损{avg_loss:,.0f}元
最大单笔盈利+{max_profit:,.0f}元
最大单笔亏损{max_loss:,.0f}元

🔄 多空统计

方向交易次数胜率总盈亏
做多 {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}元

🚪 退出原因分析

""" for reason, count in exit_reasons.items(): pct = count / total_trades * 100 html += f"" html += """
退出原因次数占比
{reason}{count}{pct:.1f}%

📝 最近10笔交易明细

""" recent_trades = trades_df.tail(10) for _, trade in recent_trades.iterrows(): pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative" html += f""" """ html += """
方向 开仓时间 平仓时间 开仓价 平仓价 盈亏金额 盈亏% 退出原因
{trade['方向']} {trade['开仓时间']} {trade['平仓时间']} {trade['开仓价']:.2f} {trade['平仓价']:.2f} {trade['盈亏金额']:+.0f} {trade['盈亏百分比']:+.2f}% {trade['退出原因']}

本报告由 cat-fly 自动交易系统生成 | 策略：30分钟K线多空双向
风险提示：历史回测不代表未来表现，投资有风险，入市需谨慎。

""" # 生成纯文本版本 text = 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} 平均持仓: {avg_hold_time*0.5:.1f}小时 【盈亏统计】 总盈利: +{total_profit:,.0f}元 总亏损: -{total_loss:,.0f}元 最大单笔盈利: +{max_profit:,.0f}元 最大单笔亏损: {max_loss:,.0f}元 【多空统计】 做多: {len(long_trades)}笔, 盈亏{long_trades['盈亏金额'].sum():+,.0f}元 做空: {len(short_trades)}笔, 盈亏{short_trades['盈亏金额'].sum():+,.0f}元 【退出原因】 {exit_reasons.to_string()} 【最近5笔交易】 {trades_df.tail(5)[['方向', '开仓时间', '平仓时间', '盈亏金额', '退出原因']].to_string(index=False)} """ return html, text # ==================== 主程序 ==================== def main(): """主程序""" print("="*80) print("🚀 cat-fly 自动交易报告系统") print("="*80) print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # 检查是否在交易时间（可选） now = datetime.now() hour = now.hour minute = now.minute time_str = f"{hour:02d}:{minute:02d}" # A股交易时间检查 is_trading_time = False if (9 <= hour <= 11) or (13 <= hour <= 15): if hour == 9 and minute < 30: is_trading_time = False elif hour == 11 and minute > 30: is_trading_time = False elif hour == 15 and minute > 0: is_trading_time = False else: is_trading_time = True print(f"当前时间: {time_str}") print(f"交易时间: {'是' if is_trading_time else '否（非交易时间也会执行）'}") # 1. 获取近2个月数据 print("\n📊 步骤1: 获取近2个月数据...") df = DataFetcher.fetch_recent_2months() if df is None: print("❌ 数据获取失败，退出") return # 2. 运行策略 print("\n📈 步骤2: 运行策略回测...") strategy = CatFlyStrategy() df, trades_df, final_capital = strategy.backtest(df) print(f"✅ 回测完成: 共{len(trades_df)}笔交易") print(f" 最终资金: {final_capital:,.0f}元") print(f" 收益率: {(final_capital/1000000-1)*100:+.2f}%") # 3. 生成报告 print("\n📝 步骤3: 生成报告...") html_report, text_report = generate_report(trades_df, final_capital) # 4. 发送邮件 print("\n📧 步骤4: 发送邮件...") subject = f"🚀 创业板50交易报告 {datetime.now().strftime('%m-%d %H:%M')} | 收益{(final_capital/1000000-1)*100:+.2f}%" # 检查邮件配置 if EMAIL_CONFIG['sender_email'] == 'your_email@qq.com': print("⚠️ 警告: 请先修改 EMAIL_CONFIG 中的邮箱配置！") print(" 配置文件位于脚本开头的 EMAIL_CONFIG 字典") print("\n📋 报告预览（前500字符）:") print(text_report[:500]) else: send_email(subject, html_report, text_report) print("\n✅ 全部完成!") print("="*80) if __name__ == "__main__": main()