cyb50-quant/cat-fly/realtime_signal.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50实时信号计算器
在任意时刻获取实时行情，计算当前信号状态
"""

import pandas as pd
import numpy as np
import akshare as ak
import sys
import warnings
warnings.filterwarnings('ignore')

sys.path.insert(0, '/root/.openclaw/workspace/cat-fly')
from cyb50_30min_dual_direction import ConfigManager, IntradayDataFetcher, DualDirectionSignalGenerator
from datetime import datetime, timedelta


def get_realtime_price():
    """获取创业板50实时价格"""
    try:
        # 使用新浪接口获取实时行情（更稳定）
        df = ak.stock_zh_index_spot_sina()
        cyb50 = df[df['代码'] == 'sz399673']
        if len(cyb50) > 0:
            return {
                'price': float(cyb50.iloc[0]['最新价']),
                'open': float(cyb50.iloc[0]['今开']),
                'high': float(cyb50.iloc[0]['最高']),
                'low': float(cyb50.iloc[0]['最低']),
                'volume': float(cyb50.iloc[0]['成交量']),
                'time': datetime.now().strftime('%H:%M:%S')
            }
    except Exception as e:
        print(f"获取实时行情失败: {e}")
    return None


def calculate_signal_at_time(current_price, last_kline_data):
    """
    基于最新价格和最后一根K线数据，估算当前信号状态
    
    参数:
        current_price: 当前实时价格
        last_kline_data: 最后一根完整K线的技术指标数据 (Series)
    """
    score = 0
    signals = []
    
    # 1. RSI估算 - 假设价格变化与RSI变化大致线性
    last_close = last_kline_data['Close']
    price_change_pct = (current_price - last_close) / last_close
    
    # 简化估算：价格每跌1%，RSI约跌2-3个点
    estimated_rsi = last_kline_data['RSI'] + price_change_pct * 250
    
    if estimated_rsi < 30:
        score += 2
        signals.append(f"RSI超卖(估{estimated_rsi:.1f})")
    elif estimated_rsi < 35:
        score += 1
        signals.append(f"RSI偏弱(估{estimated_rsi:.1f})")
    
    # 2. KDJ估算 - 价格影响J值较大
    estimated_j = last_kline_data['J'] + price_change_pct * 300
    
    if estimated_j < 0:
        score += 1
        signals.append(f"KDJ极端超卖(估J={estimated_j:.1f})")
    
    # 3. 布林带位置判断
    bb_lower = last_kline_data['BB_lower']
    bb_upper = last_kline_data['BB_upper']
    
    if current_price <= bb_lower * 1.005:
        score += 2
        signals.append("触及下轨")
    elif current_price <= bb_lower * 1.02:
        score += 1
        signals.append("接近下轨")
    
    # 4. 价格动量 - 使用实时价格与开盘/昨日收盘比较
    if price_change_pct < -0.015:
        score += 1
        signals.append(f"动量超卖({price_change_pct*100:.2f}%)")
    
    # 5. MA趋势判断（基于已有数据）
    if last_kline_data['MA6'] > last_kline_data['MA12']:
        score += 1
        signals.append("MA短期上行")
    else:
        score -= 1
        signals.append("MA下降趋势惩罚")
    
    return {
        'score': score,
        'signals': signals,
        'estimated_rsi': estimated_rsi,
        'estimated_j': estimated_j,
        'price_change_pct': price_change_pct,
        'bb_lower': bb_lower,
        'bb_upper': bb_upper,
        'current_price': current_price,
        'last_close': last_close
    }


def generate_realtime_report():
    """生成实时信号报告"""
    print("="*80)
    print("🚀 创业板50实时信号检测")
    print(f"检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*80)
    
    # 1. 获取历史30分钟K线数据（用于计算技术指标）
    print("\n📊 获取历史数据...")
    config_manager = ConfigManager('config.json')
    fetcher = IntradayDataFetcher(config_manager)
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=30)  # 取30天足够计算指标
    raw_data = fetcher.fetch_30min_data(start_date, end_date)
    
    if raw_data is None or len(raw_data) == 0:
        print("❌ 数据获取失败")
        return None
    
    # 2. 计算技术指标
    data_with_indicators = fetcher.calculate_intraday_indicators(raw_data)
    last_kline = data_with_indicators.iloc[-1]
    
    print(f"   最后一根K线: {data_with_indicators.index[-1]}")
    print(f"   收盘价: {last_kline['Close']:.2f}")
    
    # 3. 获取实时价格
    print("\n📈 获取实时行情...")
    realtime = get_realtime_price()
    
    if realtime is None:
        print("❌ 获取实时价格失败")
        return None
    
    print(f"   当前时间: {realtime['time']}")
    print(f"   当前价格: {realtime['price']:.2f}")
    print(f"   今日最高: {realtime['high']:.2f}")
    print(f"   今日最低: {realtime['low']:.2f}")
    
    # 4. 计算当前信号
    print("\n🎯 计算实时信号...")
    signal_result = calculate_signal_at_time(realtime['price'], last_kline)
    
    print(f"\n{'='*80}")
    print("实时信号评估结果")
    print(f"{'='*80}")
    print(f"当前价格: {signal_result['current_price']:.2f}")
    print(f"最后一根K线收盘: {signal_result['last_close']:.2f}")
    print(f"价格变化: {signal_result['price_change_pct']*100:+.2f}%")
    print(f"\n技术指标估算:")
    print(f"  RSI(估算): {signal_result['estimated_rsi']:.2f}")
    print(f"  KDJ J(估算): {signal_result['estimated_j']:.2f}")
    print(f"  布林带: [{signal_result['bb_lower']:.2f}, {signal_result['bb_upper']:.2f}]")
    print(f"\n信号评分: {signal_result['score']}/4 (阈值: 4分触发买入)")
    print(f"触发信号: {', '.join(signal_result['signals']) if signal_result['signals'] else '无'}")
    
    if signal_result['score'] >= 4:
        print(f"\n🟢 建议: 触发买入信号！")
    elif signal_result['score'] >= 3:
        print(f"\n🟡 建议: 接近触发，建议关注")
    else:
        print(f"\n⚪ 建议: 未触发买入信号")
    
    print(f"{'='*80}")
    
    return signal_result


if __name__ == "__main__":
    generate_realtime_report()