cyb50-quant/cat-fly/trend_report_v2.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 实时交易报告系统 (含实时数据获取)
基于趋势跟踪策略，生成当前时点报告
"""

import sys
sys.path.insert(0, '/root/.openclaw/workspace/cat-fly')
sys.path.insert(0, '/root/.openclaw/workspace/quant')

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')

# ==================== 邮件配置 ====================
EMAIL_CONFIG = {
    "smtp_server": "localhost",
    "smtp_port": 25,
    "sender_email": "catfly@erwin.wang",
    "receiver_email": "380880504@qq.com"
}

# ==================== 实时数据获取 ====================
def fetch_realtime_data():
    """获取实时数据，与历史数据合并"""
    try:
        # 使用东方财富或新浪财经接口获取最新数据
        import requests
        import json
        
        # 尝试东方财富接口
        url = "http://push2.eastmoney.com/api/qt/stock/get"
        params = {
            "secid": "0.399673",  # 创业板50
            "fields": "f43,f44,f45,f46,f47,f48,f50,f51,f52,f57,f58,f60,f107,f108,f109,f110,f111,f112,f113,f114,f115,f116,f117,f118,f119,f120,f121,f122,f123,f124,f125,f126,f127,f128,f129,f130,f131,f132,f133,f134,f135,f136,f137,f138,f139,f140,f141,f142,f143,f144,f145,f146,f147,f148,f149,f150,f151,f152,f153,f154,f155,f156,f157,f158,f159,f160,f161,f162,f163,f164,f165,f166,f167,f168,f169,f170,f171,f172,f173,f174,f175,f176,f177,f178,f179,f180,f181,f182,f183,f184,f185,f186,f187,f188,f189,f190,f191,f192,f193,f194,f195,f196,f197,f198,f199,f200"
        }
        
        response = requests.get(url, params=params, timeout=10)
        data = response.json()
        
        if 'data' in data and data['data']:
            d = data['data']
            # 解析实时数据
            current_price = float(d.get('f43', 0)) / 100  # 当前价
            open_price = float(d.get('f46', 0)) / 100     # 开盘价
            high_price = float(d.get('f44', 0)) / 100     # 最高价
            low_price = float(d.get('f45', 0)) / 100      # 最低价
            volume = int(d.get('f47', 0))                 # 成交量
            
            today = datetime.now().strftime('%Y-%m-%d')
            
            return {
                'date': today,
                'open': open_price,
                'high': high_price,
                'low': low_price,
                'close': current_price,
                'volume': volume
            }
    except Exception as e:
        print(f"实时数据获取失败: {e}")
        return None


# ==================== 趋势跟踪策略 ====================
class TrendTrackingStrategy:
    """趋势跟踪策略 - 实时计算版"""
    
    def __init__(self):
        self.data = None
        
    def load_and_merge_data(self, csv_file='cyb50_baostock.csv'):
        """加载历史数据并合并实时数据"""
        try:
            # 加载历史数据
            df = pd.read_csv(f'/root/.openclaw/workspace/quant/{csv_file}')
            df['date'] = pd.to_datetime(df['date'])
            df = df.set_index('date').sort_index()
            
            # 转换数据类型
            for col in ['open', 'high', 'low', 'close', 'volume']:
                df[col] = pd.to_numeric(df[col], errors='coerce')
            
            # 获取最新历史数据日期
            last_hist_date = df.index[-1]
            today = pd.Timestamp.now().normalize()
            
            print(f"历史数据最新日期: {last_hist_date.date()}")
            print(f"当前日期: {today.date()}")
            
            # 如果历史数据不是最新的，尝试获取实时数据
            if last_hist_date < today:
                print("历史数据不是最新，尝试获取实时数据...")
                realtime = fetch_realtime_data()
                
                if realtime and pd.Timestamp(realtime['date']) > last_hist_date:
                    # 创建新数据行
                    new_row = pd.DataFrame([realtime]).set_index('date')
                    new_row.index = pd.to_datetime(new_row.index)
                    
                    # 合并数据
                    df = pd.concat([df, new_row])
                    print(f"✅ 已合并实时数据: {realtime['date']}")
                else:
                    print("⚠️ 未获取到更新的实时数据，使用历史数据")
            else:
                print("✅ 历史数据已是最新")
            
            self.data = df
            print(f"数据范围: {df.index[0].date()} ~ {df.index[-1].date()}")
            return True
            
        except Exception as e:
            print(f"数据加载失败: {e}")
            return False
    
    def calculate_indicators(self):
        """计算技术指标"""
        df = self.data.copy()
        
        # 均线
        df['ma10'] = df['close'].rolling(window=10).mean()
        df['ma30'] = df['close'].rolling(window=30).mean()
        
        # 20日高低点
        df['high_20'] = df['high'].rolling(window=20).max()
        df['low_20'] = df['low'].rolling(window=20).min()
        
        # 10日涨幅
        df['ret_10'] = df['close'].pct_change(periods=10)
        
        # ATR
        tr1 = df['high'] - df['low']
        tr2 = abs(df['high'] - df['close'].shift(1))
        tr3 = abs(df['low'] - df['close'].shift(1))
        df['tr'] = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        df['atr'] = df['tr'].rolling(window=20).mean()
        
        self.data = df
        return df
    
    def generate_signals(self):
        """生成交易信号"""
        df = self.data
        
        # 买入条件
        buy_cond = (
            (df['close'] > df['ma10']) & 
            (df['ma10'] > df['ma30']) & 
            (df['close'] >= df['high_20'] * 0.995) & 
            (df['ret_10'] > 0.02)
        )
        
        # 卖出条件
        sell_cond = (
            (df['close'] < df['ma30']) | 
            (df['close'] <= df['low_20'] * 1.005)
        )
        
        df['signal'] = 0
        df.loc[buy_cond, 'signal'] = 1
        df.loc[sell_cond, 'signal'] = -1
        
        return df
    
    def backtest(self, initial_capital=1000000):
        """回测计算"""
        df = self.generate_signals()
        
        position = 0
        entry_price = 0
        peak_price = 0
        capital = initial_capital
        trades = []
        
        for i in range(30, len(df)):
            date = df.index[i]
            price = df['close'].iloc[i]
            signal = df['signal'].iloc[i]
            
            # 移动止损检查
            if position > 0:
                if price > peak_price:
                    peak_price = price
                if price < peak_price * 0.90:  # 10%回撤止损
                    signal = -1
            
            # 执行交易
            if signal == 1 and position == 0:
                position = 1
                entry_price = price
                peak_price = price
                trades.append({
                    'date': date,
                    'action': 'BUY',
                    'price': price,
                    'capital': capital
                })
            
            elif signal == -1 and position == 1:
                pnl = (price / entry_price - 1) * capital
                capital += pnl
                position = 0
                trades.append({
                    'date': date,
                    'action': 'SELL',
                    'price': price,
                    'capital': capital,
                    'pnl': pnl,
                    'return_pct': (price / entry_price - 1) * 100
                })
        
        # 计算当前持仓
        current_position = position
        current_price = df['close'].iloc[-1]
        
        if position == 1:
            unrealized_pnl = (current_price / entry_price - 1) * capital
            total_value = capital + unrealized_pnl
        else:
            total_value = capital
        
        total_return = (total_value / initial_capital - 1) * 100
        
        return {
            'trades': trades,
            'current_position': current_position,
            'current_price': current_price,
            'entry_price': entry_price if position == 1 else None,
            'capital': capital,
            'total_value': total_value,
            'total_return': total_return,
            'trade_count': len([t for t in trades if t['action'] == 'SELL']),
            'data_end_date': df.index[-1].strftime('%Y-%m-%d')
        }
    
    def get_recent_indicators(self, days=20):
        """获取近N天指标详情"""
        df = self.data.tail(days).copy()
        
        indicators = []
        for date, row in df.iterrows():
            indicators.append({
                'date': date.strftime('%Y-%m-%d'),
                'close': round(row['close'], 2),
                'ma10': round(row['ma10'], 2) if not pd.isna(row['ma10']) else '-',
                'ma30': round(row['ma30'], 2) if not pd.isna(row['ma30']) else '-',
                'high_20': round(row['high_20'], 2) if not pd.isna(row['high_20']) else '-',
                'ret_10': f"{row['ret_10']*100:.2f}%" if not pd.isna(row['ret_10']) else '-',
                'signal': '买入' if row['signal'] == 1 else ('卖出' if row['signal'] == -1 else '持有'),
                'atr': round(row['atr'], 2) if not pd.isna(row['atr']) else '-'
            })
        
        return indicators
    
    def get_recent_trades(self, n=20):
        """获取近N次交易详情"""
        result = self.backtest()
        trades = result['trades'][-n:]
        return trades


def generate_report():
    """生成完整报告"""
    strategy = TrendTrackingStrategy()
    
    if not strategy.load_and_merge_data():
        return None, None, None
    
    strategy.calculate_indicators()
    result = strategy.backtest()
    
    # 获取近期数据
    recent_indicators = strategy.get_recent_indicators(20)
    recent_trades = strategy.get_recent_trades(20)
    
    # 生成HTML报告
    html = f"""
    <html>
    <head>
        <meta charset="utf-8">
        <style>
            body {{ font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }}
            .container {{ max-width: 1200px; margin: 0 auto; background: white; padding: 20px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
            h1 {{ color: #333; border-bottom: 3px solid #007bff; padding-bottom: 10px; }}
            h2 {{ color: #555; margin-top: 30px; border-left: 4px solid #007bff; padding-left: 10px; }}
            .summary {{ background: #f8f9fa; padding: 15px; border-radius: 5px; margin: 20px 0; }}
            .metric {{ display: inline-block; margin: 10px 20px 10px 0; }}
            .metric-label {{ color: #666; font-size: 12px; }}
            .metric-value {{ font-size: 24px; font-weight: bold; color: #333; }}
            .positive {{ color: #28a745; }}
            .negative {{ color: #dc3545; }}
            table {{ width: 100%; border-collapse: collapse; margin: 20px 0; font-size: 13px; }}
            th {{ background: #007bff; color: white; padding: 10px; text-align: left; }}
            td {{ padding: 8px 10px; border-bottom: 1px solid #ddd; }}
            tr:nth-child(even) {{ background: #f8f9fa; }}
            tr:hover {{ background: #e9ecef; }}
            .position-yes {{ color: #28a745; font-weight: bold; }}
            .position-no {{ color: #666; }}
            .buy {{ color: #28a745; font-weight: bold; }}
            .sell {{ color: #dc3545; font-weight: bold; }}
            .data-info {{ background: #e7f3ff; padding: 10px; border-radius: 5px; margin: 10px 0; color: #004085; }}
        </style>
    </head>
    <body>
        <div class="container">
            <h1>🚀 创业板50趋势跟踪策略报告</h1>
            <p style="color: #666;">生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
            
            <div class="data-info">
                <strong>数据更新:</strong> 已合并历史数据与实时数据，最新数据日期: {result['data_end_date']}
            </div>
            
            <div class="summary">
                <h2>📊 总体绩效</h2>
                <div class="metric">
                    <div class="metric-label">当前持仓</div>
                    <div class="metric-value {'position-yes' if result['current_position'] == 1 else 'position-no'}">
                        {'持有中' if result['current_position'] == 1 else '空仓'}
                    </div>
                </div>
                <div class="metric">
                    <div class="metric-label">当前价格</div>
                    <div class="metric-value">{result['current_price']:.2f}</div>
                </div>
                <div class="metric">
                    <div class="metric-label">累计收益率</div>
                    <div class="metric-value {'positive' if result['total_return'] >= 0 else 'negative'}">
                        {result['total_return']:+.2f}%
                    </div>
                </div>
                <div class="metric">
                    <div class="metric-label">总资产</div>
                    <div class="metric-value">{result['total_value']:,.0f}元</div>
                </div>
                <div class="metric">
                    <div class="metric-label">交易次数</div>
                    <div class="metric-value">{result['trade_count']}</div>
                </div>
            </div>
    """
    
    # 持仓详情
    if result['current_position'] == 1:
        unrealized = (result['current_price'] / result['entry_price'] - 1) * 100
        html += f"""
            <div class="summary" style="background: #e8f5e9;">
                <h2>📈 持仓详情</h2>
                <p><strong>入场价格:</strong> {result['entry_price']:.2f} 元</p>
                <p><strong>当前浮盈:</strong> <span class="{'positive' if unrealized >= 0 else 'negative'}">{unrealized:+.2f}%</span></p>
            </div>
        """
    
    # 近20天指标
    html += """
            <h2>📅 近20天指标详情</h2>
            <table>
                <tr>
                    <th>日期</th>
                    <th>收盘价</th>
                    <th>MA10</th>
                    <th>MA30</th>
                    <th>20日高</th>
                    <th>10日涨幅</th>
                    <th>信号</th>
                    <th>ATR</th>
                </tr>
    """
    
    for ind in recent_indicators:
        signal_class = 'buy' if ind['signal'] == '买入' else ('sell' if ind['signal'] == '卖出' else '')
        html += f"""
                <tr>
                    <td>{ind['date']}</td>
                    <td>{ind['close']}</td>
                    <td>{ind['ma10']}</td>
                    <td>{ind['ma30']}</td>
                    <td>{ind['high_20']}</td>
                    <td>{ind['ret_10']}</td>
                    <td class="{signal_class}">{ind['signal']}</td>
                    <td>{ind['atr']}</td>
                </tr>
        """
    
    html += """
            </table>
            
            <h2>💼 近20次交易详情</h2>
            <table>
                <tr>
                    <th>日期</th>
                    <th>操作</th>
                    <th>价格</th>
                    <th>盈亏金额</th>
                    <th>盈亏比例</th>
                    <th>总资产</th>
                </tr>
    """
    
    for trade in recent_trades:
        action_class = 'buy' if trade['action'] == 'BUY' else 'sell'
        pnl = trade.get('pnl', 0)
        ret = trade.get('return_pct', 0)
        html += f"""
                <tr>
                    <td>{trade['date'].strftime('%Y-%m-%d') if isinstance(trade['date'], pd.Timestamp) else trade['date']}</td>
                    <td class="{action_class}">{trade['action']}</td>
                    <td>{trade['price']:.2f}</td>
                    <td>{f'{pnl:+,.0f}元' if 'pnl' in trade else '-'}</td>
                    <td class="{'positive' if ret >= 0 else 'negative'}">{f'{ret:+.2f}%' if 'return_pct' in trade else '-'}</td>
                    <td>{trade['capital']:,.0f}元</td>
                </tr>
        """
    
    html += """
            </table>
            
            <div style="margin-top: 30px; padding: 15px; background: #fff3cd; border-radius: 5px; color: #856404;">
                <strong>策略说明:</strong><br>
                • 买入条件: 价格>MA10>MA30 且 突破20日高×0.995 且 10日涨幅>2%<br>
                • 卖出条件: 跌破MA30 或 创20日新低 或 回撤10%止损<br>
                • 数据更新: 已合并历史数据与实时市场数据
            </div>
        </div>
    </body>
    </html>
    """
    
    # 生成文本报告
    text = f"""
创业板50趋势跟踪策略报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
数据最新日期: {result['data_end_date']}

【总体绩效】
当前持仓: {'持有中' if result['current_position'] == 1 else '空仓'}
当前价格: {result['current_price']:.2f}元
累计收益率: {result['total_return']:+.2f}%
总资产: {result['total_value']:,.0f}元
交易次数: {result['trade_count']}

【近20天指标】
日期        收盘价    MA10     MA30     20日高   10日涨幅  信号  ATR
"""
    
    for ind in recent_indicators:
        text += f"{ind['date']}  {ind['close']:>8}  {ind['ma10']:>8}  {ind['ma30']:>8}  {ind['high_20']:>8}  {ind['ret_10']:>8}  {ind['signal']:>4}  {ind['atr']:>6}\n"
    
    text += "\n【近20次交易】\n"
    text += "日期        操作    价格      盈亏金额     盈亏比例    总资产\n"
    
    for trade in recent_trades:
        pnl_str = f"{trade.get('pnl', 0):+,.0f}元" if 'pnl' in trade else '-'
        ret_str = f"{trade.get('return_pct', 0):+.2f}%" if 'return_pct' in trade else '-'
        date_str = trade['date'].strftime('%Y-%m-%d') if isinstance(trade['date'], pd.Timestamp) else trade['date']
        text += f"{date_str}  {trade['action']:>6}  {trade['price']:>8.2f}  {pnl_str:>12}  {ret_str:>10}  {trade['capital']:>12,.0f}元\n"
    
    return html, text, result


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


def main():
    """主程序"""
    print("="*60)
    print("🚀 创业板50趋势跟踪实时报告 (含实时数据)")
    print("="*60)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 生成报告
    print("\n📊 加载数据并生成报告中...")
    html, text, result = generate_report()
    
    if html is None:
        print("❌ 报告生成失败")
        return
    
    print(f"✅ 报告生成完成")
    print(f"   数据最新日期: {result['data_end_date']}")
    print(f"   当前持仓: {'持有中' if result['current_position'] == 1 else '空仓'}")
    print(f"   累计收益: {result['total_return']:+.2f}%")
    print(f"   交易次数: {result['trade_count']}")
    
    # 发送邮件
    print("\n📧 发送邮件...")
    position_status = "持仓" if result['current_position'] == 1 else "空仓"
    subject = f"🚀 创业板50趋势报告 {datetime.now().strftime('%m-%d %H:%M')} | {position_status} | 收益{result['total_return']:+.2f}%"
    send_email(subject, html, text)
    
    print("\n✅ 全部完成!")
    print("="*60)


if __name__ == "__main__":
    main()