#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 创业板50指数 - 自动化交易报告系统 (完整版) 基于原版 cyb50_30min_intraday_reversal.py 策略 功能： 1. 获取近2个月实时数据 2. 运行完整策略回测（做多翻转策略） 3. 生成详细报告 4. 发送邮件通知 """ import pandas as pd import numpy as np import akshare as ak import warnings import os import smtplib import ssl import requests import json 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": "localhost", "smtp_port": 25, "sender_email": "catfly@openclaw.local", "sender_password": "", "receiver_email": "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'] = EMAIL_CONFIG['receiver_email'] 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_email'], msg.as_string() ) print(f"✅ 邮件发送成功: {subject}") return True except Exception as e: print(f"❌ 邮件发送失败: {e}") return False # ==================== 数据获取 ==================== class DataFetcher: """数据获取类 - 双数据源""" @staticmethod def fetch_recent_2months(): """获取近2个月数据""" end_date = datetime.now() start_date = end_date - timedelta(days=70) 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线...") 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() # 计算基础指标 df['Returns'] = df['Close'].pct_change() df['High_Low_Pct'] = (df['High'] - df['Low']) / df['Close'].shift(1) df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open'] # 保留近2个月 backtest_start = end_date - timedelta(days=60) df_backtest = df[df.index >= backtest_start].copy() print(f"✅ 东方财富: {len(df_backtest)}条K线 ({df_backtest.index[0]} ~ {df_backtest.index[-1]})") return df_backtest except Exception as e: print(f"❌ 东方财富: {e}") return None @staticmethod def _fetch_sina_data(start_date, end_date): """新浪财经数据源""" try: print("[数据源2] 新浪财经30分钟K线...") 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', 'Referer': 'https://finance.sina.com.cn/' } response = requests.get(url, headers=headers, timeout=15) json_start = response.text.find('[') json_end = response.text.rfind(']') + 1 json_str = response.text[json_start:json_end] data_dict = json.loads(json_str) if data_dict and isinstance(data_dict, list): data_list = [] for item in data_dict: 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)) }) df = pd.DataFrame(data_list) df['DateTime'] = pd.to_datetime(df['DateTime']) df = df.set_index('DateTime').sort_index() # 计算基础指标 df['Returns'] = df['Close'].pct_change() df['High_Low_Pct'] = (df['High'] - df['Low']) / df['Close'].shift(1) df['Close_Open_Pct'] = (df['Close'] - df['Open']) / df['Open'] # 保留近2个月 backtest_start = end_date - timedelta(days=60) df_backtest = df[df.index >= backtest_start].copy() print(f"✅ 新浪财经: {len(df_backtest)}条K线 ({df_backtest.index[0]} ~ {df_backtest.index[-1]})") return df_backtest except Exception as e: print(f"❌ 新浪财经: {e}") return None # ==================== 策略类（完整版） ==================== class CatFlyStrategy: """cat-fly完整策略 - 日内翻转做多策略""" 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.015, 'max_hold_bars': 16, 'min_reversal_score': 4 # 原版阈值是4 } self.initial_capital = self.config['initial_capital'] def calculate_indicators(self, df): """计算完整技术指标（与原版一致）""" print("计算技术指标...") # 短期移动平均线 df['MA6'] = df['Close'].rolling(window=6).mean() df['MA12'] = df['Close'].rolling(window=12).mean() df['MA24'] = df['Close'].rolling(window=24).mean() # 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['BB_middle'] = df['Close'].rolling(window=20).mean() bb_std = df['Close'].rolling(window=20).std() df['BB_upper'] = df['BB_middle'] + (bb_std * 2) df['BB_lower'] = df['BB_middle'] - (bb_std * 2) df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle'] # MACD exp1 = df['Close'].ewm(span=12, adjust=False).mean() exp2 = df['Close'].ewm(span=26, adjust=False).mean() df['MACD'] = exp1 - exp2 df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean() df['MACD_hist'] = df['MACD'] - df['MACD_signal'] # KDJ low_9 = df['Low'].rolling(window=9).min() high_9 = df['High'].rolling(window=9).max() rsv = (df['Close'] - low_9) / (high_9 - low_9) * 100 df['K'] = rsv.ewm(com=2, adjust=False).mean() df['D'] = df['K'].ewm(com=2, adjust=False).mean() df['J'] = 3 * df['K'] - 2 * df['D'] # ATR high_low = df['High'] - df['Low'] high_close = abs(df['High'] - df['Close'].shift()) low_close = abs(df['Low'] - df['Close'].shift()) true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) df['ATR'] = true_range.rolling(window=14).mean() df['ATR_Pct'] = df['ATR'] / df['Close'] # 动量指标 df['Momentum'] = df['Close'] / df['Close'].shift(4) - 1 # 成交量变化 df['Volume_MA'] = df['Volume'].rolling(window=12).mean() df['Volume_Ratio'] = df['Volume'] / df['Volume_MA'] # 价格动量 df['Price_Momentum'] = (df['Close'] - df['Close'].shift(6)) / df['Close'].shift(6) print("技术指标计算完成") return df def generate_signals(self, df): """生成日内翻转信号（完整版）""" print("生成交易信号...") df = self.calculate_indicators(df) signals = [] signal_count = 0 # 从第24个周期开始（需要足够历史数据） for i in range(24, len(df)): current_bar = df.iloc[i] current_time = df.index[i] # 跳过非交易时间 if hasattr(current_time, 'hour'): hour = current_time.hour if hour < 9 or hour > 15: continue # 计算翻转信号分数 reversal_score = 0 reversal_signals = [] # 1. RSI超卖翻转 if current_bar['RSI'] < 30: reversal_score += 2 reversal_signals.append("RSI超卖") elif current_bar['RSI'] < 35: reversal_score += 1 reversal_signals.append("RSI偏弱") # 2. KDJ超卖翻转 if current_bar['K'] < 20 and current_bar['D'] < 20: reversal_score += 2 reversal_signals.append("KDJ超卖") elif current_bar['J'] < 0: reversal_score += 2 reversal_signals.append("KDJ极端超卖") # 3. MACD金叉或改善 if current_bar['MACD_hist'] > 0 and df.iloc[i-1]['MACD_hist'] <= 0: reversal_score += 2 reversal_signals.append("MACD金叉") elif current_bar['MACD_hist'] > df.iloc[i-1]['MACD_hist']: reversal_score += 1 reversal_signals.append("MACD改善") # 4. 价格触及布林带下轨 if current_bar['Close'] <= current_bar['BB_lower'] * 1.005: reversal_score += 2 reversal_signals.append("触及下轨") elif current_bar['Close'] <= current_bar['BB_lower'] * 1.01: reversal_score += 1 reversal_signals.append("接近下轨") # 5. 连续下跌后的反转 recent_returns = df.iloc[i-6:i]['Returns'] if recent_returns.min() < -0.015: consecutive_decline = sum(recent_returns < 0) if consecutive_decline >= 4: reversal_score += 2 reversal_signals.append("连续下跌反转") # 6. 价格动量超卖 if current_bar['Price_Momentum'] < -0.02: reversal_score += 1 reversal_signals.append("动量超卖") # 7. 成交量配合 if current_bar['Volume_Ratio'] > 1.2: reversal_score += 1 reversal_signals.append("放量配合") # 8. 日内低位 try: daily_data = df[df.index.date == current_time.date()] if len(daily_data) > 0: daily_high = daily_data['High'].max() daily_low = daily_data['Low'].min() daily_range = daily_high - daily_low if daily_range > 0: position_in_day = (current_bar['Close'] - daily_low) / daily_range if position_in_day < 0.3: reversal_score += 1 reversal_signals.append("日内低位") except: pass # 记录信号 df.loc[df.index[i], 'Reversal_Score'] = reversal_score df.loc[df.index[i], 'Reversal_Signals'] = ', '.join(reversal_signals) # 生成买入信号（阈值4分） if reversal_score >= self.config['min_reversal_score']: df.loc[df.index[i], 'Signal'] = 1 df.loc[df.index[i], 'Signal_Type'] = '做多翻转' signal_count += 1 else: df.loc[df.index[i], 'Signal'] = 0 df.loc[df.index[i], 'Signal_Type'] = '' print(f"信号生成完成: 共{signal_count}个翻转信号") return df def backtest(self, df): """回测执行""" df = self.generate_signals(df) trades = [] capital = self.initial_capital position = 0 entry_price = 0 entry_time = None entry_signals = "" holding_bars = 0 for i in range(24, len(df)): current_bar = df.iloc[i] price = current_bar['Close'] current_time = df.index[i] # 检查是否在交易时间 if hasattr(current_time, 'hour'): hour = current_time.hour minute = current_time.minute if hour == 11 and minute >= 30: # 午休前不新开仓 pass if hour == 15: # 收盘前不新开仓 pass # 无持仓时检查开仓 if position == 0: if current_bar.get('Signal', 0) == 1: position_value = capital * self.config['position_size_pct'] position_size = int(position_value / price) if position_size > 0: position = position_size entry_price = price entry_time = current_time entry_signals = current_bar.get('Reversal_Signals', '') holding_bars = 0 # 有持仓时检查平仓 elif position > 0: holding_bars += 1 exit_signal = False exit_reason = "" # 止损 stop_loss_price = entry_price * (1 - self.config['stop_loss_pct']) take_profit_price = entry_price * (1 + self.config['take_profit_pct']) if price <= stop_loss_price: exit_signal = True exit_reason = f"止损({self.config['stop_loss_pct']*100:.1f}%)" # 止盈 elif price >= take_profit_price: exit_signal = True exit_reason = f"止盈({self.config['take_profit_pct']*100:.1f}%)" # 时间止损 elif holding_bars >= self.config['max_hold_bars']: exit_signal = True exit_reason = f"时间止损({holding_bars}周期)" # RSI超买平仓 elif current_bar['RSI'] > 70: exit_signal = True exit_reason = "RSI超买平仓" # 执行平仓 if exit_signal: pnl = (price - entry_price) * position pnl_pct = (price - entry_price) / entry_price * 100 capital += pnl trades.append({ '方向': '做多', '开仓时间': entry_time, '平仓时间': current_time, '开仓价': entry_price, '平仓价': price, '持仓数量': position, '盈亏金额': pnl, '盈亏百分比': pnl_pct, '退出原因': exit_reason, '持仓周期': holding_bars, '信号详情': entry_signals, '平仓后资金': 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): """生成详细报告""" total_return = (final_capital - initial_capital) / initial_capital * 100 total_trades = len(trades_df) if total_trades == 0: html = f"""

🚀 创业板50交易报告

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

数据区间: 近2个月

近2个月无交易信号触发

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

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

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

说明：策略在指定期间内未找到符合条件的翻转信号（需RSI<30、KDJ超卖、MACD金叉等多重条件同时满足）

""" text = f""" 创业板50交易报告 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 数据区间: 近2个月 近2个月无交易信号触发 初始资金: {initial_capital:,.0f}元 最终资金: {final_capital:,.0f}元 收益率: {total_return:+.2f}% 说明：策略在指定期间内未找到符合条件的翻转信号 """ return html, text # 有交易数据时的报告... winning_trades = trades_df[trades_df['盈亏金额'] > 0] losing_trades = trades_df[trades_df['盈亏金额'] < 0] win_rate = len(winning_trades) / total_trades * 100 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 # 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}

📝 最近10笔交易明细

""" for _, trade in trades_df.tail(10).iterrows(): pnl_class = "positive" if trade['盈亏金额'] >= 0 else "negative" html += f""" """ html += "
开仓时间平仓时间开仓价平仓价 盈亏盈亏%退出原因信号
{trade['开仓时间']}{trade['平仓时间']} {trade['开仓价']:.2f}{trade['平仓价']:.2f} {trade['盈亏金额']:+.0f} {trade['盈亏百分比']:+.2f}% {trade['退出原因']}{trade['信号详情'][:20]}...
" # 纯文本报告 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} 【最近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')}") # 1. 获取数据 print("\n📊 步骤1: 获取近2个月实时数据...") try: df = DataFetcher.fetch_recent_2months() except Exception as e: print(f"❌ 数据获取失败: {e}") 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}% | {len(trades_df)}笔交易" send_email(subject, html_report, text_report) print("\n✅ 全部完成!") print("="*80) if __name__ == "__main__": main()