dragon

dragon/MyTT.py

@@ -0,0 +1,185 @@
+# MyTT 麦语言-通达信-同花顺指标实现    https://github.com/mpquant/MyTT
+# V2.1 2021-6-6 新增 BARSLAST函数
+# V2.2 2021-6-8 新增 SLOPE,FORCAST线性回归，和回归预测函数
+  
+import numpy as np; import pandas as pd
+
+#------------------ 0级：核心工具函数 --------------------------------------------      
+def RD(N,D=3):   return np.round(N,D)        #四舍五入取3位小数 
+def RET(S,N=1):  return np.array(S)[-N]      #返回序列倒数第N个值,默认返回最后一个
+def ABS(S):      return np.abs(S)            #返回N的绝对值
+def MAX(S1,S2):  return np.maximum(S1,S2)    #序列max
+def MIN(S1,S2):  return np.minimum(S1,S2)    #序列min
+         
+def MA(S,N):           #求序列的N日平均值，返回序列                    
+    return pd.Series(S).rolling(N).mean().values
+
+def REF(S, N=1):       #对序列整体下移动N,返回序列(shift后会产生NAN)    
+    return pd.Series(S).shift(N).values  
+
+def DIFF(S, N=1):      #前一个值减后一个值,前面会产生nan 
+    return pd.Series(S).diff(N)  #np.diff(S)直接删除nan，会少一行
+
+def STD(S,N):           #求序列的N日标准差，返回序列    
+    return  pd.Series(S).rolling(N).std(ddof=0).values     
+
+def IF(S_BOOL,S_TRUE,S_FALSE):          #序列布尔判断 res=S_TRUE if S_BOOL==True  else  S_FALSE
+    return np.where(S_BOOL, S_TRUE, S_FALSE)
+
+def SUM(S, N):                          #对序列求N天累计和，返回序列         
+    return pd.Series(S).rolling(N).sum().values
+
+def HHV(S,N):                           # HHV(C, 5)  # 最近5天收盘最高价        
+    return pd.Series(S).rolling(N).max().values
+
+def LLV(S,N):                           # LLV(C, 5)  # 最近5天收盘最低价     
+    return pd.Series(S).rolling(N).min().values
+
+def EMA(S,N):         #指数移动平均,为了精度 S>4*N  EMA至少需要120周期       
+    return pd.Series(S).ewm(span=N, adjust=False).mean().values    
+
+def SMA(S, N, M=1):   #中国式的SMA,至少需要120周期才精确         
+    K = pd.Series(S).rolling(N).mean()    #先求出平均值 (下面如果有不用循环的办法，能提高性能，望告知)
+    for i in range(N+1, len(S)):  K[i] = (M * S[i] + (N -M) * K[i-1]) / N  # 因为要取K[i-1]，所以 range(N+1, len(S))        
+    return K
+
+def AVEDEV(S,N):      #平均绝对偏差  (序列与其平均值的绝对差的平均值)   
+    avedev=pd.Series(S).rolling(N).apply(lambda x: (np.abs(x - x.mean())).mean())    
+    return avedev.values
+
+def SLOPE(S,N,RS=False):               #返S序列N周期回线性回归斜率 (默认只返回斜率,不返回整个直线序列)
+    M=pd.Series(S[-N:]);   poly = np.polyfit(M.index, M.values,deg=1);    Y=np.polyval(poly, M.index); 
+    if RS: return Y[1]-Y[0],Y
+    return Y[1]-Y[0]
+
+  
+#------------------   1级：应用层函数(通过0级核心函数实现） ----------------------------------
+def COUNT(S_BOOL, N):                  # COUNT(CLOSE>O, N):  最近N天满足S_BOO的天数  True的天数
+    return SUM(S_BOOL,N)    
+
+def EVERY(S_BOOL, N):                  # EVERY(CLOSE>O, 5)   最近N天是否都是True
+    R=SUM(S_BOOL, N)
+    return  IF(R==N, True, False)
+  
+def LAST(S_BOOL, A, B):                #从前A日到前B日一直满足S_BOOL条件   
+    if A<B: A=B                        #要求A>B    例：LAST(CLOSE>OPEN,5,3)  5天前到3天前是否都收阳线     
+    return S_BOOL[-A:-B].sum()==(A-B)  #返回单个布尔值    
+
+def EXIST(S_BOOL, N=5):                # EXIST(CLOSE>3010, N=5)  n日内是否存在一天大于3000点
+    R=SUM(S_BOOL,N)    
+    return IF(R>0, True ,False)
+
+def BARSLAST(S_BOOL):                  #上一次条件成立到当前的周期  
+    M=np.argwhere(S_BOOL);             # BARSLAST(CLOSE/REF(CLOSE)>=1.1) 上一次涨停到今天的天数
+    return len(S_BOOL)-int(M[-1])-1  if M.size>0 else -1
+
+def FORCAST(S,N):                      #返S序列N周期回线性回归后的预测值
+    K,Y=SLOPE(S,N,RS=True)
+    return Y[-1]+K
+  
+def CROSS(S1,S2):                      #判断穿越 CROSS(MA(C,5),MA(C,10))               
+    CROSS_BOOL=IF(S1>S2, True ,False)   
+    return COUNT(CROSS_BOOL>0,2)==1    #上穿：昨天0 今天1   下穿：昨天1 今天0
+
+
+
+#------------------   2级：技术指标函数(全部通过0级，1级函数实现） ------------------------------
+def MACD(CLOSE,SHORT=12,LONG=26,M=9):            # EMA的关系，S取120日，和雪球小数点2位相同
+    DIF = EMA(CLOSE,SHORT)-EMA(CLOSE,LONG);  
+    DEA = EMA(DIF,M);      MACD=(DIF-DEA)*2
+    return RD(DIF),RD(DEA),RD(MACD)
+
+def KDJ(CLOSE,HIGH,LOW, N=9,M1=3,M2=3):         # KDJ指标
+    RSV = (CLOSE - LLV(LOW, N)) / (HHV(HIGH, N) - LLV(LOW, N)) * 100
+    K = EMA(RSV, (M1*2-1));    D = EMA(K,(M2*2-1));        J=K*3-D*2
+    return K, D, J
+
+def RSI(CLOSE, N=24):      
+    DIF = CLOSE-REF(CLOSE,1) 
+    return RD(SMA(MAX(DIF,0), N) / SMA(ABS(DIF), N) * 100)  
+
+def WR(CLOSE, HIGH, LOW, N=10, N1=6):            #W&R 威廉指标
+    WR = (HHV(HIGH, N) - CLOSE) / (HHV(HIGH, N) - LLV(LOW, N)) * 100
+    WR1 = (HHV(HIGH, N1) - CLOSE) / (HHV(HIGH, N1) - LLV(LOW, N1)) * 100
+    return RD(WR), RD(WR1)
+
+def BIAS(CLOSE,L1=6, L2=12, L3=24):              # BIAS乖离率
+    BIAS1 = (CLOSE - MA(CLOSE, L1)) / MA(CLOSE, L1) * 100
+    BIAS2 = (CLOSE - MA(CLOSE, L2)) / MA(CLOSE, L2) * 100
+    BIAS3 = (CLOSE - MA(CLOSE, L3)) / MA(CLOSE, L3) * 100
+    return RD(BIAS1), RD(BIAS2), RD(BIAS3)
+
+def BOLL(CLOSE,N=20, P=2):                       #BOLL指标，布林带    
+    MID = MA(CLOSE, N); 
+    UPPER = MID + STD(CLOSE, N) * P
+    LOWER = MID - STD(CLOSE, N) * P
+    return RD(UPPER), RD(MID), RD(LOWER)    
+
+def PSY(CLOSE,N=12, M=6):  
+    PSY=COUNT(CLOSE>REF(CLOSE,1),N)/N*100
+    PSYMA=MA(PSY,M)
+    return RD(PSY),RD(PSYMA)
+
+def CCI(CLOSE,HIGH,LOW,N=14):  
+    TP=(HIGH+LOW+CLOSE)/3
+    return (TP-MA(TP,N))/(0.015*AVEDEV(TP,N))
+        
+def ATR(CLOSE,HIGH,LOW, N=20):                    #真实波动N日平均值
+    TR = MAX(MAX((HIGH - LOW), ABS(REF(CLOSE, 1) - HIGH)), ABS(REF(CLOSE, 1) - LOW))
+    return MA(TR, N)
+
+def BBI(CLOSE,M1=3,M2=6,M3=12,M4=20):             #BBI多空指标   
+    return (MA(CLOSE,M1)+MA(CLOSE,M2)+MA(CLOSE,M3)+MA(CLOSE,M4))/4    
+
+def DMI(CLOSE,HIGH,LOW,M1=14,M2=6):               #动向指标：结果和同花顺，通达信完全一致
+    TR = SUM(MAX(MAX(HIGH - LOW, ABS(HIGH - REF(CLOSE, 1))), ABS(LOW - REF(CLOSE, 1))), M1)
+    HD = HIGH - REF(HIGH, 1);     LD = REF(LOW, 1) - LOW
+    DMP = SUM(IF((HD > 0) & (HD > LD), HD, 0), M1)
+    DMM = SUM(IF((LD > 0) & (LD > HD), LD, 0), M1)
+    PDI = DMP * 100 / TR;         MDI = DMM * 100 / TR
+    ADX = MA(ABS(MDI - PDI) / (PDI + MDI) * 100, M2)
+    ADXR = (ADX + REF(ADX, M2)) / 2
+    return PDI, MDI, ADX, ADXR  
+
+def TAQ(HIGH,LOW,N):                              #唐安奇通道交易指标，大道至简，能穿越牛熊
+    UP=HHV(HIGH,N);    DOWN=LLV(LOW,N);    MID=(UP+DOWN)/2
+    return UP,MID,DOWN
+
+def TRIX(CLOSE,M1=12, M2=20):                      #三重指数平滑平均线
+    TR = EMA(EMA(EMA(CLOSE, M1), M1), M1)
+    TRIX = (TR - REF(TR, 1)) / REF(TR, 1) * 100
+    TRMA = MA(TRIX, M2)
+    return TRIX, TRMA
+
+def VR(CLOSE,VOL,M1=26):                           #VR容量比率
+    LC = REF(CLOSE, 1)
+    return SUM(IF(CLOSE > LC, VOL, 0), M1) / SUM(IF(CLOSE <= LC, VOL, 0), M1) * 100
+
+def EMV(HIGH,LOW,VOL,N=14,M=9):                     #简易波动指标 
+    VOLUME=MA(VOL,N)/VOL;       MID=100*(HIGH+LOW-REF(HIGH+LOW,1))/(HIGH+LOW)
+    EMV=MA(MID*VOLUME*(HIGH-LOW)/MA(HIGH-LOW,N),N);    MAEMV=MA(EMV,M)
+    return EMV,MAEMV
+
+
+def DPO(CLOSE,M1=20, M2=10, M3=6):                  #区间震荡线
+    DPO = CLOSE - REF(MA(CLOSE, M1), M2);    MADPO = MA(DPO, M3)
+    return DPO, MADPO
+
+def BRAR(OPEN,CLOSE,HIGH,LOW,M1=26):                 #BRAR-ARBR 情绪指标  
+    AR = SUM(HIGH - OPEN, M1) / SUM(OPEN - LOW, M1) * 100
+    BR = SUM(MAX(0, HIGH - REF(CLOSE, 1)), M1) / SUM(MAX(0, REF(CLOSE, 1) - LOW), M1) * 100
+    return AR, BR
+
+def DMA(CLOSE,N1=10,N2=50,M=10):                     #平行线差指标  
+    DIF=MA(CLOSE,N1)-MA(CLOSE,N2);    DIFMA=MA(DIF,M)
+    return DIF,DIFMA
+
+def MTM(CLOSE,N=12,M=6):                             #动量指标
+    MTM=CLOSE-REF(CLOSE,N);         MTMMA=MA(MTM,M)
+    return MTM,MTMMA
+
+def ROC(CLOSE,N=12,M=6):                             #变动率指标
+    ROC=100*(CLOSE-REF(CLOSE,N))/REF(CLOSE,N);    MAROC=MA(ROC,M)
+    return ROC,MAROC  
+  
+  #望大家能提交更多指标和函数  https://github.com/mpquant/MyTT

dragon/data_fetcher_v2.py

@@ -0,0 +1,320 @@
+import akshare as ak
+import pandas as pd
+import numpy as np
+from datetime import datetime, timedelta
+from typing import Optional, Union, List, Tuple
+import time
+
+class DataFetcherV2:
+    """
+    数据获取类V2 - 基于用户提供的优化方法
+    使用ak.stock_zh_index_daily获取更可靠的数据
+    """
+    
+    def __init__(self):
+        self.cache = {}
+        self.cache_expiry = {}
+        self.cache_duration = 3600  # 缓存1小时
+        
+    def _get_cache_key(self, symbol: str, start_date: str, end_date: str) -> str:
+        """生成缓存键"""
+        return f"{symbol}_{start_date}_{end_date}"
+    
+    def _is_cache_valid(self, cache_key: str) -> bool:
+        """检查缓存是否有效"""
+        if cache_key not in self.cache_expiry:
+            return False
+        return time.time() < self.cache_expiry[cache_key]
+    
+    def _set_cache(self, cache_key: str, data: pd.DataFrame):
+        """设置缓存"""
+        self.cache[cache_key] = data
+        self.cache_expiry[cache_key] = time.time() + self.cache_duration
+    
+    def _format_index_code(self, symbol: str) -> str:
+        """
+        格式化指数代码为akshare标准格式
+        例如: 399673 -> sz399673
+        """
+        symbol = symbol.strip()
+        
+        # 移除可能的前缀
+        if '.' in symbol:
+            code, exchange = symbol.split('.')
+            symbol = code
+        
+        # 确保是6位代码
+        if len(symbol) == 6:
+            # 判断交易所并添加前缀
+            if symbol.startswith(('00', '30')):  # 深交所
+                return f"sz{symbol}"
+            elif symbol.startswith(('60', '68')):  # 上交所
+                return f"sh{symbol}"
+            else:
+                # 其他交易所默认使用sz
+                return f"sz{symbol}"
+        
+        # 如果已经是格式化好的代码
+        return symbol.lower()
+    
+    def fetch_index_data_v2(self, 
+                            symbol: str, 
+                            start_date: str = "2018-01-01", 
+                            end_date: Optional[str] = None) -> pd.DataFrame:
+        """
+        使用优化的方法获取指数数据
+        
+        Args:
+            symbol: 指数代码，支持多种格式
+            start_date: 开始日期，默认2018-01-01
+            end_date: 结束日期，默认为当前日期
+            
+        Returns:
+            包含OHLCV数据的DataFrame，索引为日期
+        """
+        if end_date is None:
+            end_date = datetime.now().strftime('%Y-%m-%d')
+        
+        cache_key = self._get_cache_key(symbol, start_date, end_date)
+        
+        if self._is_cache_valid(cache_key):
+            return self.cache[cache_key].copy()
+        
+        try:
+            # 格式化指数代码
+            formatted_code = self._format_index_code(symbol)
+            print(f"正在获取指数 {formatted_code} 的日线级别历史数据...")
+            
+            # 使用akshare获取日线数据（用户提供的优化方法）
+            all_data_df = ak.stock_zh_index_daily(symbol=formatted_code)
+            
+            if all_data_df.empty:
+                print(f"Warning: No data found for index {symbol}")
+                return pd.DataFrame()
+            
+            # 处理日期列
+            all_data_df['date'] = pd.to_datetime(all_data_df['date'])
+            all_data_df.set_index('date', inplace=True)
+            
+            # 筛选日期范围
+            start_datetime = pd.to_datetime(start_date)
+            end_datetime = pd.to_datetime(end_date)
+            
+            # 先筛选出指定日期之后的数据
+            filtered_df = all_data_df[all_data_df.index >= start_datetime]
+            filtered_df = filtered_df[filtered_df.index <= end_datetime]
+            
+            if filtered_df.empty:
+                print(f"Warning: No data found for {symbol} in date range {start_date} to {end_date}")
+                return pd.DataFrame()
+            
+            # 标准化列名
+            filtered_df = self._standardize_columns(filtered_df)
+            
+            print(f"数据获取成功，期间为 {filtered_df.index[0]} 到 {filtered_df.index[-1]}")
+            print(f"获取数据量: {len(filtered_df)} 条")
+            
+            # 缓存数据
+            self._set_cache(cache_key, filtered_df)
+            return filtered_df.copy()
+            
+        except Exception as e:
+            print(f"Error fetching index data for {symbol}: {str(e)}")
+            return pd.DataFrame()
+    
+    def fetch_stock_data_v2(self, 
+                           symbol: str, 
+                           start_date: str = "2018-01-01", 
+                           end_date: Optional[str] = None) -> pd.DataFrame:
+        """
+        获取股票数据（如果akshare支持的话）
+        """
+        if end_date is None:
+            end_date = datetime.now().strftime('%Y-%m-%d')
+        
+        # 目前主要支持指数数据，股票数据作为占位符
+        print(f"Warning: Stock data fetching not fully implemented, trying as index: {symbol}")
+        return self.fetch_index_data_v2(symbol, start_date, end_date)
+    
+    def _standardize_columns(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        标准化DataFrame列名和索引
+        """
+        # 确保索引是日期类型
+        if not isinstance(df.index, pd.DatetimeIndex):
+            try:
+                df.index = pd.to_datetime(df.index)
+            except:
+                pass
+        
+        # 创建列名映射（中英文对照）
+        column_mapping = {
+            '开盘': 'open', 
+            '收盘': 'close',
+            '最高': 'high',
+            '最低': 'low',
+            '成交量': 'volume',
+            '成交额': 'amount',
+            '涨跌幅': 'change_pct',
+            '涨跌额': 'change',
+            '振幅': 'amplitude',
+            '换手率': 'turnover'
+        }
+        
+        # 重命名列
+        df = df.rename(columns=column_mapping)
+        
+        # 确保必要的列存在
+        required_columns = ['open', 'high', 'low', 'close', 'volume']
+        for col in required_columns:
+            if col not in df.columns:
+                # 尝试从中文列名获取
+                chinese_map = {
+                    'open': '开盘', 'high': '最高', 'low': '最低', 
+                    'close': '收盘', 'volume': '成交量'
+                }
+                if chinese_map[col] in df.columns:
+                    df[col] = df[chinese_map[col]]
+                else:
+                    print(f"Warning: Missing column {col}, filling with NaN")
+                    df[col] = np.nan
+        
+        # 选择并排序列
+        result_columns = [col for col in required_columns if col in df.columns]
+        other_columns = [col for col in df.columns if col not in required_columns]
+        df = df[result_columns + other_columns]
+        
+        # 确保数值列是数值类型
+        numeric_columns = ['open', 'high', 'low', 'close', 'volume']
+        for col in numeric_columns:
+            if col in df.columns:
+                df[col] = pd.to_numeric(df[col], errors='coerce')
+        
+        # 删除包含NaN的行（可选）
+        # df = df.dropna(subset=['close'])
+        
+        return df
+    
+    def get_data_by_date_range(self, 
+                              symbol: str, 
+                              start_date: str, 
+                              end_date: str) -> pd.DataFrame:
+        """
+        获取指定日期范围的数据（主要接口）
+        
+        Args:
+            symbol: 指数代码
+            start_date: 开始日期 'YYYY-MM-DD'
+            end_date: 结束日期 'YYYY-MM-DD'
+            
+        Returns:
+            DataFrame with date index and OHLCV columns
+        """
+        return self.fetch_index_data_v2(symbol, start_date, end_date)
+    
+    def clear_cache(self):
+        """清除缓存"""
+        self.cache.clear()
+        self.cache_expiry.clear()
+
+
+class DataManagerV2:
+    """
+    数据管理类V2 - 基于优化的数据获取方式
+    """
+    
+    def __init__(self, data_fetcher: DataFetcherV2):
+        self.data_fetcher = data_fetcher
+        self.data_cache = {}
+        self.market_info = {}
+        
+    def load_data(self, 
+                  symbol: str, 
+                  start_date: str, 
+                  end_date: str) -> pd.DataFrame:
+        """
+        加载数据并缓存
+        
+        Args:
+            symbol: 指数代码
+            start_date: 开始日期
+            end_date: 结束日期
+            
+        Returns:
+            DataFrame with standardized format
+        """
+        cache_key = f"{symbol}_{start_date}_{end_date}"
+        
+        if cache_key not in self.data_cache:
+            print(f"Loading data for {symbol} from {start_date} to {end_date}...")
+            data = self.data_fetcher.get_data_by_date_range(symbol, start_date, end_date)
+            
+            if data.empty:
+                print(f"Warning: Empty data returned for {symbol}")
+            else:
+                print(f"Successfully loaded {len(data)} bars for {symbol}")
+                # 存储市场信息
+                self.market_info[symbol] = {
+                    'start_date': data.index[0],
+                    'end_date': data.index[-1],
+                    'total_bars': len(data),
+                    'has_data': True
+                }
+            
+            self.data_cache[cache_key] = data
+        
+        return self.data_cache[cache_key].copy()
+    
+    def get_market_info(self, symbol: str) -> dict:
+        """获取市场信息"""
+        return self.market_info.get(symbol, {})
+    
+    def get_available_symbols(self) -> list:
+        """获取已加载的标的列表"""
+        return list(self.market_info.keys())
+    
+    def get_current_data(self, 
+                        symbol: str, 
+                        current_date: str,
+                        window_size: int = 100) -> pd.DataFrame:
+        """
+        获取当前日期之前的数据窗口（支持日期索引格式）
+        """
+        # 查找对应的数据缓存
+        cache_key = None
+        for key, data in self.data_cache.items():
+            if not data.empty and key.startswith(symbol):
+                cache_key = key
+                break
+        
+        if cache_key is None:
+            return pd.DataFrame()
+        
+        data = self.data_cache[cache_key]
+        current_datetime = pd.to_datetime(current_date)
+        
+        # 支持两种数据格式：日期索引或date列
+        if isinstance(data.index, pd.DatetimeIndex):
+            # 日期索引格式
+            historical_data = data[data.index <= current_datetime].copy()
+            result_data = historical_data.tail(window_size).reset_index(drop=True)
+        else:
+            # date列格式
+            historical_data = data[data['date'] <= current_datetime].copy()
+            result_data = historical_data.tail(window_size).reset_index(drop=True)
+        
+        if len(result_data) == 0:
+            return pd.DataFrame()
+        
+        return result_data
+    
+    def print_data_summary(self):
+        """打印数据摘要"""
+        print("\n" + "="*70)
+        print("DATA MANAGER SUMMARY")
+        print("="*70)
+        for symbol, info in self.market_info.items():
+            print(f"{symbol}:")
+            print(f"  Period: {info['start_date']} to {info['end_date']}")
+            print(f"  Total bars: {info['total_bars']}")
+        print("="*70)

dragon/t0_signal_analyzer.py

@@ -0,0 +1,537 @@
+#!/usr/bin/env python3
+# coding=utf-8
+"""
+T0交易信号分析器
+支持日内T0交易，打印最近10天的买卖交易信号及原因
+"""
+import sys
+import io
+import pandas as pd
+import numpy as np
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional
+from dataclasses import dataclass
+
+# 设置标准输出为UTF-8编码
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
+
+from data_fetcher_v2 import DataFetcherV2, DataManagerV2
+import MyTT
+
+# 尝试导入requests用于实时数据
+try:
+    import requests
+    REQUESTS_AVAILABLE = True
+except ImportError:
+    REQUESTS_AVAILABLE = False
+    print("⚠️  requests未安装，无法获取实时数据")
+
+
+@dataclass
+class TradingSignal:
+    """交易信号记录"""
+    date: datetime
+    symbol: str
+    close_price: float
+    y0: float
+    y1: float
+    y2: float
+    y3: float
+    h1: float
+    h2: float
+    a1: float
+    b1: float
+    cross_y0_y1: bool
+    cross_y1_y0: bool
+    signal_type: str  # 'BUY', 'SELL', 'NONE'
+    reason: str  # 信号原因或不触发原因
+    position_status: str  # 当前持仓状态
+
+
+class T0StrategyAnalyzer:
+    """T0策略分析器 - 支持日内交易"""
+
+    def __init__(self):
+        self.maPeriod = 26
+        self.stdPeriod = 150
+        self.stdRange = 1
+        self.symbol = '399673'
+        self.period = max(self.maPeriod, self.stdPeriod, self.stdRange) + 1
+
+        # T0相关变量
+        self.t0_bought_today = {}  # 当日买入的股票 {symbol: volume}
+        self.t0_available_to_sell = {}  # T0可用卖出数量
+
+        # 信号记录
+        self.signals: List[TradingSignal] = []
+
+    def fetch_realtime_data(self) -> Optional[pd.Series]:
+        """获取实时数据 - 使用新浪API（策略计算不需要成交量）"""
+        if not REQUESTS_AVAILABLE:
+            print("⚠️  requests模块不可用")
+            return None
+
+        try:
+            print("正在获取实时行情数据...")
+
+            # 新浪指数实时行情API
+            code = self.symbol.replace('sz', '').replace('sh', '')
+            # 判断前缀 - 399xxx是深交所指数，使用sz前缀
+            if self.symbol.startswith('sz') or (len(code) == 6 and code.startswith('3')):
+                prefix = 'sz'
+            elif self.symbol.startswith('sh') or (len(code) == 6 and code.startswith('0')):
+                prefix = 'sh'
+            else:
+                prefix = 'sz'
+
+            sina_url = f'http://hq.sinajs.cn/list={prefix}{code}'
+            headers = {
+                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
+                'Referer': 'http://finance.sina.com.cn'
+            }
+            response = requests.get(sina_url, headers=headers, timeout=10)
+
+            if response.status_code == 200:
+                # 修复编码问题
+                response.encoding = 'gbk'
+                content = response.text
+
+                if content and '"' in content:
+                    # 解析新浪数据格式
+                    data_str = content.split('"')[1]
+                    parts = data_str.split(',')
+                    # 新浪数据格式: 名称,今开,昨收,现价,最高,最低,...
+                    if len(parts) >= 6:
+                        try:
+                            name = parts[0]
+                            open_price = float(parts[1])
+                            close_prev = float(parts[2])
+                            current_price = float(parts[3])
+                            high_price = float(parts[4])
+                            low_price = float(parts[5])
+
+                            # 数据有效性检查
+                            if current_price > 0:
+                                change_pct = ((current_price - close_prev) / close_prev) * 100
+                                print(f"✅ {name} 现价={current_price:.2f} ({change_pct:+.2f}%), "
+                                      f"今开={open_price:.2f}, 最高={high_price:.2f}, 最低={low_price:.2f}")
+
+                                return pd.Series({
+                                    'date': datetime.now(),
+                                    'open': open_price,
+                                    'high': high_price,
+                                    'low': low_price,
+                                    'close': current_price,
+                                    'volume': 0  # 策略不需要成交量
+                                })
+                        except (ValueError, IndexError) as e:
+                            print(f"⚠️  数据解析失败: {e}")
+
+        except requests.exceptions.Timeout:
+            print("⚠️  请求超时")
+        except requests.exceptions.ConnectionError:
+            print("⚠️  网络连接失败")
+        except Exception as e:
+            print(f"⚠️  获取实时数据失败: {e}")
+
+        return None
+
+    def _format_index_code(self, symbol: str) -> str:
+        """格式化指数代码"""
+        symbol = symbol.strip()
+        if len(symbol) == 6:
+            if symbol.startswith(('00', '30')):
+                return f"sz{symbol}"
+            elif symbol.startswith(('60', '68')):
+                return f"sh{symbol}"
+        return symbol.lower()
+
+    def fetch_data(self, start_date: str, end_date: str) -> pd.DataFrame:
+        """获取数据"""
+        print(f"获取历史数据: {start_date} 至 {end_date}")
+        data_fetcher = DataFetcherV2()
+        data = data_fetcher.fetch_index_data_v2(
+            symbol=self.symbol,
+            start_date=start_date,
+            end_date=end_date
+        )
+
+        if data.empty:
+            print("❌ 数据获取失败")
+            return pd.DataFrame()
+
+        # 检查数据的实际日期范围
+        actual_start = data.index[0].strftime('%Y-%m-%d')
+        actual_end = data.index[-1].strftime('%Y-%m-%d')
+
+        print(f"✅ 获取到 {len(data)} 条历史数据")
+        print(f"   数据范围: {actual_start} 至 {actual_end}")
+
+        # 检查是否有今天的数据
+        today = datetime.now()
+        today_str = today.strftime('%Y-%m-%d')
+        actual_end_dt = pd.to_datetime(actual_end)
+
+        if actual_end_dt.date() < today.date():
+            print(f"\n⚠️  历史数据未包含今天({today_str})，最后交易日为 {actual_end}")
+
+            # 检查今天是否是交易日（排除周末）
+            if today.weekday() < 5:  # 0-4是周一到周五
+                print(f"今天是{['周一', '周二', '周三', '周四', '周五'][today.weekday()]}，尝试获取实时数据...")
+
+                # 尝试获取实时数据
+                realtime = self.fetch_realtime_data()
+                if realtime is not None:
+                    # 将实时数据添加到历史数据
+                    realtime_df = pd.DataFrame([realtime])
+                    realtime_df = realtime_df.set_index('date')
+                    realtime_df.index = pd.to_datetime(realtime_df.index)
+
+                    # 使用concat合并数据
+                    data = pd.concat([data, realtime_df], ignore_index=False)
+                    # 重新排序索引
+                    data = data.sort_index()
+                    print(f"   ✅ 已添加今日({today_str})实时数据，总数据量: {len(data)} 条")
+                else:
+                    print(f"   ⚠️  未能获取实时数据，将使用历史数据分析")
+            else:
+                print(f"   ℹ️  今天是周末，非交易日")
+
+        return data
+
+    def calculate_indicators(self, data: pd.DataFrame) -> dict:
+        """计算技术指标"""
+        if data.empty or len(data) < self.period:
+            return None
+
+        close = data['close'].values
+        low = data['low'].values
+        high = data['high'].values
+
+        # 计算EMA
+        H1_5 = MyTT.EMA(close, 8)
+        H2_5 = MyTT.EMA(H1_5, 20)
+
+        # 计算CROSS
+        try:
+            H1H2_CROSS = MyTT.CROSS(H1_5, H2_5)
+            H2H1_CROSS = MyTT.CROSS(H2_5, H1_5)
+        except:
+            H1H2_CROSS = (H1_5 > H2_5) & (np.roll(H1_5, 1) <= np.roll(H2_5, 1))
+            H2H1_CROSS = (H2_5 > H1_5) & (np.roll(H2_5, 1) <= np.roll(H1_5, 1))
+
+        # KDJ相关计算
+        rsv = (close - MyTT.LLV(low, 7)) / (MyTT.HHV(high, 7) - MyTT.LLV(low, 7)) * 100
+        rsv = np.nan_to_num(rsv)
+        Y0 = MyTT.SMA(rsv, 3, 1)
+        Y0 = np.nan_to_num(Y0)
+        Y1 = MyTT.SMA(Y0, 3, 1)
+        Y1 = np.nan_to_num(Y1)
+
+        try:
+            CROSS_Y0_Y1 = MyTT.CROSS(Y0, Y1)
+            CROSS_Y1_Y0 = MyTT.CROSS(Y1, Y0)
+        except:
+            CROSS_Y0_Y1 = (Y0 > Y1) & (np.roll(Y0, 1) <= np.roll(Y1, 1))
+            CROSS_Y1_Y0 = (Y1 > Y0) & (np.roll(Y1, 1) <= np.roll(Y0, 1))
+
+        RSV1 = (close - MyTT.LLV(low, 38)) / (MyTT.HHV(high, 38) - MyTT.LLV(low, 38)) * 100
+        RSV1 = np.nan_to_num(RSV1)
+        Y2 = MyTT.SMA(RSV1, 5, 1)
+        Y2 = np.nan_to_num(Y2)
+        Y3 = MyTT.SMA(Y2, 10, 1)
+        Y3 = np.nan_to_num(Y3)
+
+        a1 = (H1_5[-1] - H2_5[-1]) / ((H1_5[-1] + H2_5[-1]) / 2) if (H1_5[-1] + H2_5[-1]) / 2 != 0 else 0
+        b1 = (Y2[-1] - Y3[-1]) / 100
+
+        def safe_bool(value):
+            if isinstance(value, (bool, np.bool_)):
+                return bool(value)
+            elif isinstance(value, (int, float)):
+                return bool(value)
+            else:
+                try:
+                    return bool(value[-1] if hasattr(value, '__iter__') else value)
+                except:
+                    return False
+
+        return {
+            'H1_5': H1_5[-1],
+            'H2_5': H2_5[-1],
+            'Y0': Y0[-1],
+            'Y1': Y1[-1],
+            'Y2': Y2[-1],
+            'Y3': Y3[-1],
+            'a1': a1,
+            'b1': b1,
+            'cross_y0_y1': safe_bool(CROSS_Y0_Y1[-1]),
+            'cross_y1_y0': safe_bool(CROSS_Y1_Y0[-1]),
+        }
+
+    def analyze_signal(
+        self,
+        indicators: dict,
+        has_position: bool,
+        current_date: datetime,
+        close_price: float
+    ) -> TradingSignal:
+        """分析交易信号"""
+
+        y0 = indicators['Y0']
+        y1 = indicators['Y1']
+        y2 = indicators['Y2']
+        y3 = indicators['Y3']
+        h1 = indicators['H1_5']
+        h2 = indicators['H2_5']
+        a1 = indicators['a1']
+        b1 = indicators['b1']
+        cross_y0_y1 = indicators['cross_y0_y1']
+        cross_y1_y0 = indicators['cross_y1_y0']
+
+        signal_type = 'NONE'
+        reason = ''
+
+        # 1. 首先检查交叉信号条件
+        if not cross_y0_y1 or not cross_y1_y0:
+            # 交叉条件不满足
+            cross_status = []
+            if not cross_y0_y1:
+                cross_status.append("Y0未上穿Y1")
+            if not cross_y1_y0:
+                cross_status.append("Y1未上穿Y0")
+            reason = f"交叉条件不满足: {', '.join(cross_status)}"
+
+        # 2. 交叉条件满足后的买入判断
+        elif y0 > y1 and b1 > 0 and (a1 > -0.02 or a1 < 0.02):
+            # 检查排除条件
+            if a1 < -0.04:
+                reason = f"买入信号排除: a1({a1:.4f}) < -0.04 (趋势过弱)"
+            elif b1 < -0.17:
+                reason = f"买入信号排除: b1({b1:.4f}) < -0.17 (动量过弱)"
+            elif not has_position:
+                signal_type = 'BUY'
+                reason = f"买入信号触发: Y0({y0:.2f})>Y1({y1:.2f}), b1({b1:.4f})>0"
+            else:
+                reason = f"已有持仓，不追加买入: Y0({y0:.2f})>Y1({y1:.2f}), b1({b1:.4f})>0"
+
+        # 3. 交叉条件满足后的卖出判断
+        elif y0 <= y1 and (a1 > -0.02 or a1 < 0.02):
+            # 检查排除条件
+            if a1 > 0.05:
+                reason = f"卖出信号排除: a1({a1:.4f}) > 0.05 (趋势过强)"
+            elif has_position:
+                signal_type = 'SELL'
+                reason = f"卖出信号触发: Y0({y0:.2f})<=Y1({y1:.2f}), a1({a1:.4f})条件满足"
+            else:
+                reason = f"无持仓，无法卖出: Y0({y0:.2f})<=Y1({y1:.2f})"
+
+        # 4. 其他情况
+        else:
+            conditions = []
+            if y0 <= y1:
+                conditions.append(f"Y0({y0:.2f})<=Y1({y1:.2f})")
+            else:
+                conditions.append(f"Y0({y0:.2f})>Y1({y1:.2f})")
+
+            if b1 <= 0:
+                conditions.append(f"b1({b1:.4f})<=0")
+
+            if not (a1 > -0.02 or a1 < 0.02):
+                conditions.append(f"a1({a1:.4f})不在[-0.02, 0.02]范围")
+
+            reason = f"不满足任何交易条件: {', '.join(conditions)}"
+
+        return TradingSignal(
+            date=current_date,
+            symbol=self.symbol,
+            close_price=close_price,
+            y0=y0,
+            y1=y1,
+            y2=y2,
+            y3=y3,
+            h1=h1,
+            h2=h2,
+            a1=a1,
+            b1=b1,
+            cross_y0_y1=cross_y0_y1,
+            cross_y1_y0=cross_y1_y0,
+            signal_type=signal_type,
+            reason=reason,
+            position_status="持仓" if has_position else "空仓"
+        )
+
+    def run_t0_backtest(
+        self,
+        start_date: str,
+        end_date: str,
+        initial_cash: float = 1000000
+    ):
+        """运行T0回测并分析信号"""
+
+        print("=" * 80)
+        print("T0策略信号分析")
+        print("=" * 80)
+
+        # 获取数据
+        data = self.fetch_data(start_date, end_date)
+        if data.empty:
+            return
+
+        # 模拟持仓和现金
+        cash = initial_cash
+        position_volume = 0
+        position_cost = 0
+
+        # 滑动窗口分析
+        window_size = self.period
+
+        print(f"\n开始分析交易信号...")
+        print("-" * 80)
+
+        for i in range(window_size, len(data)):
+            current_date = data.index[i]
+            current_data = data.iloc[:i+1]
+
+            # 检查是否有足够数据
+            if len(current_data) < self.period:
+                continue
+
+            # 计算指标
+            indicators = self.calculate_indicators(current_data)
+            if indicators is None:
+                continue
+
+            close_price = data.iloc[i]['close']
+
+            # 分析信号
+            has_position = position_volume > 0
+            signal = self.analyze_signal(
+                indicators,
+                has_position,
+                current_date,
+                close_price
+            )
+
+            self.signals.append(signal)
+
+            # T0交易模拟
+            if signal.signal_type == 'BUY':
+                # 计算可买入数量
+                volume = int(cash / close_price)
+                volume = max(100, (volume // 100) * 100)
+
+                if volume > 0:
+                    cost = volume * close_price
+                    cash -= cost
+                    # T0: 更新持仓成本和数量（使用加权平均）
+                    if position_volume > 0:
+                        total_cost = position_cost * position_volume + cost
+                        position_volume += volume
+                        position_cost = total_cost / position_volume
+                    else:
+                        position_volume = volume
+                        position_cost = close_price
+
+            elif signal.signal_type == 'SELL':
+                if position_volume > 0:
+                    # T0: 卖出全部持仓
+                    proceeds = position_volume * close_price
+                    cash += proceeds
+                    position_volume = 0
+                    position_cost = 0
+
+        # 打印最近10天的信号
+        self.print_recent_signals(days=10)
+
+    def print_recent_signals(self, days: int = 10):
+        """打印最近N天的交易信号"""
+
+        print("\n" + "=" * 140)
+        print(f"最近 {days} 个交易日信号分析")
+        print("=" * 140)
+
+        recent_signals = self.signals[-days:] if len(self.signals) >= days else self.signals
+
+        for signal in recent_signals:
+            # 打印日期和价格
+            print(f"\n📅 {signal.date.strftime('%Y-%m-%d')} | 收盘价: {signal.close_price:.2f} | 持仓: {signal.position_status}")
+
+            # 打印指标值
+            print(f"   指标: Y0={signal.y0:.2f} Y1={signal.y1:.2f} Y2={signal.y2:.2f} Y3={signal.y3:.2f} | "
+                  f"H1={signal.h1:.2f} H2={signal.h2:.2f} | a1={signal.a1:.4f} b1={signal.b1:.4f}")
+
+            # 打印交叉状态
+            cross_str = ""
+            if signal.cross_y0_y1:
+                cross_str += "Y0↑Y1 "
+            if signal.cross_y1_y0:
+                cross_str += "Y1↑Y0 "
+            print(f"   交叉: {cross_str if cross_str else '无交叉'}")
+
+            # 打印信号
+            if signal.signal_type == 'BUY':
+                print(f"   🟢 买入信号: {signal.reason}")
+            elif signal.signal_type == 'SELL':
+                print(f"   🔴 卖出信号: {signal.reason}")
+            else:
+                print(f"   ⚪ 无信号: {signal.reason}")
+
+        print("\n" + "=" * 140)
+
+    def export_signals_to_csv(self, filename: str = "t0_signals.csv"):
+        """导出信号到CSV"""
+        if not self.signals:
+            print("没有信号可导出")
+            return
+
+        data = []
+        for signal in self.signals:
+            data.append({
+                'date': signal.date.strftime('%Y-%m-%d'),
+                'symbol': signal.symbol,
+                'close_price': signal.close_price,
+                'y0': signal.y0,
+                'y1': signal.y1,
+                'y2': signal.y2,
+                'y3': signal.y3,
+                'h1': signal.h1,
+                'h2': signal.h2,
+                'a1': signal.a1,
+                'b1': signal.b1,
+                'cross_y0_y1': signal.cross_y0_y1,
+                'cross_y1_y0': signal.cross_y1_y0,
+                'signal_type': signal.signal_type,
+                'reason': signal.reason,
+                'position_status': signal.position_status
+            })
+
+        df = pd.DataFrame(data)
+        df.to_csv(filename, index=False, encoding='utf-8-sig')
+        print(f"✅ 信号已导出到: {filename}")
+
+
+def main():
+    """主函数"""
+    # 配置 - 扩大时间范围以确保有足够数据
+    END_DATE = datetime.now().strftime('%Y-%m-%d')
+    START_DATE = (datetime.now() - timedelta(days=365*2)).strftime('%Y-%m-%d')  # 获取2年数据
+
+    print(f"分析期间: {START_DATE} 至 {END_DATE}")
+
+    # 创建分析器
+    analyzer = T0StrategyAnalyzer()
+
+    # 运行分析
+    analyzer.run_t0_backtest(
+        start_date=START_DATE,
+        end_date=END_DATE,
+        initial_cash=1000000
+    )
+
+    # 导出信号
+    analyzer.export_signals_to_csv("t0_signals.csv")
+
+
+if __name__ == '__main__':
+    main()