#!/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') # ==================== 邮件配置 ==================== # 使用本地Postfix SMTP服务器发送 EMAIL_CONFIG = { "smtp_server": "localhost", # 本地Postfix服务器 "smtp_port": 25, # SMTP端口 "sender_email": "catfly@openclaw.local", # 发件人邮箱 "sender_password": "", # 本地SMTP无需密码 "receiver_email": "380880504@qq.com" # 收件人邮箱 } def send_email(subject, html_content, text_content=""): """发送邮件 - 使用本地Postfix""" 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) # 发送邮件 - 本地Postfix无需SSL和认证 with smtplib.SMTP(EMAIL_CONFIG['smtp_server'], EMAIL_CONFIG['smtp_port']) as server: 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')}") # 尝试东方财富数据源 df = DataFetcher._fetch_eastmoney_data(start_date, end_date) if df is not None: return df # 东方财富失败，尝试新浪财经 print("⚠️ 东方财富数据源失败，尝试新浪财经...") df = DataFetcher._fetch_sina_data(start_date, end_date) if df is not None: return df # 所有数据源都失败 raise Exception("无法获取实时数据，东方财富和新浪财经均失败。请检查网络连接或稍后重试。") @staticmethod def _fetch_eastmoney_data(start_date, end_date): """从东方财富获取数据""" try: print("[数据源1] 正在使用东方财富30分钟K线接口...") # 使用东方财富接口获取30分钟K线 df = ak.index_zh_a_hist_min_em(symbol="399673", period="30") if df is not None and not df.empty and len(df) >= 50: # 标准化列名 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]}") # 检查数据时效性 latest_time = df_backtest.index[-1] time_delay = end_date - latest_time print(f" 数据延迟: {time_delay}") return df_backtest else: print(f"⚠️ 东方财富数据不足: {len(df) if df is not None else 0}条") return None except Exception as e: print(f"❌ 东方财富数据源失败: {e}") return None @staticmethod def _fetch_sina_data(start_date, end_date): """从新浪财经获取数据""" try: print("[数据源2] 正在使用新浪财经30分钟K线接口...") import requests import json import re symbol = "sz399673" url = f"https://quotes.sina.cn/cn/api/jsonp_v2.php/var_{symbol}_30_/CN_MarketDataService.getKLineData?symbol={symbol}&scale=30&ma=no&datalen=1023" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Referer': 'https://finance.sina.com.cn/' } response = requests.get(url, headers=headers, timeout=15) response_text = response.text # 解析JSONP响应 json_start = response_text.find('[') json_end = response_text.rfind(']') + 1 if json_start >= 0 and json_end > json_start: json_str = response_text[json_start:json_end] else: raise Exception("无法解析JSONP响应") data_dict = json.loads(json_str) if data_dict and isinstance(data_dict, list) and len(data_dict) > 0: data_list = [] for item in data_dict: try: data_list.append({ 'datetime': item.get('day'), 'open': float(item.get('open', 0)), 'high': float(item.get('high', 0)), 'low': float(item.get('low', 0)), 'close': float(item.get('close', 0)), 'volume': float(item.get('volume', 0)) }) except Exception: continue if data_list: df = pd.DataFrame(data_list) 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 else: print("❌ 新浪财经数据解析失败") return None else: print("❌ 新浪财经返回数据格式错误") return None 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}%" send_email(subject, html_report, text_report) print("\n✅ 全部完成!") print("="*80) if __name__ == "__main__": main()