from flask import Flask, render_template, jsonify, request import ccxt import pandas as pd from datetime import datetime, timedelta import sys import os import matplotlib matplotlib.use('Agg') # 设置使用非GUI后端，必须在导入pyplot之前设置 import matplotlib.pyplot as plt import io import base64 import time import traceback from pytz import timezone import talib.abstract as ta import numpy as np # 添加父目录到系统路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ChanLun import ChanLun from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE from cn_stock_data import ChinaStockData # 添加买卖点枚举类型 class TRADE_POINT_TYPE: BUY1 = 1 # 一类买点 BUY2 = 2 # 二类买点 BUY3 = 3 # 三类买点 SELL1 = -1 # 一类卖点 SELL2 = -2 # 二类卖点 SELL3 = -3 # 三类卖点 app = Flask(__name__) # 初始化交易所 exchange = ccxt.binance({ 'enableRateLimit': True, }) # 初始化A股数据获取器 china_stock = ChinaStockData() # 时间周期映射 TIMEFRAMES = { '1m': '1分钟', '3m': '3分钟', '5m': '5分钟', '15m': '15分钟', '30m': '30分钟', '1h': '1小时', '2h': '2小时', '4h': '4小时', '8h': '8小时', '12h': '12小时', '1d': '日线', '3d': '3日线', '1w': '周线', '1M': '月线', } # 常见交易对 SYMBOLS = [ 'SOL/USDT:USDT', 'BTC/USDT:USDT', 'ETH/USDT:USDT', 'BNB/USDT:USDT', 'XRP/USDT:USDT', 'ADA/USDT:USDT', 'DOGE/USDT:USDT', 'AVAX/USDT:USDT', 'DOT/USDT:USDT', 'MATIC/USDT:USDT' ] # A股热门股票 A_STOCK_SYMBOLS = china_stock.get_popular_stocks() def detect_symbol_type(symbol): """检测交易对类型：crypto 或 a_stock""" if '/' in symbol and 'USDT' in symbol: return 'crypto' elif len(symbol) == 6 and symbol.isdigit(): return 'a_stock' else: return 'unknown' def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): """获取K线数据，支持加密货币和A股""" symbol_type = detect_symbol_type(symbol) if symbol_type == 'crypto': return get_crypto_kl_data(symbol, timeframe, limit, start_time, end_time) elif symbol_type == 'a_stock': return get_a_stock_kl_data(symbol, timeframe, limit, start_time, end_time) else: return None def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): """获取加密货币K线数据，支持分页加载确保获取指定时间范围内的所有数据""" try: # 初始化参数 since = None if start_time: try: since = int(start_time) except ValueError: pass # 结束时间处理 until = None if end_time: try: until = int(end_time) except ValueError: pass # 根据时间周期调整每次请求的数据量 batch_size = 1000 # 默认批次大小 if timeframe in ['1m', '3m', '5m']: batch_size = 500 # 分钟级数据减少批次大小 elif timeframe in ['15m', '30m', '1h']: batch_size = 1000 else: batch_size = 1500 # 日线及以上可以获取更多 batch_size = 1000 # 默认批次大小 # 初始化存储所有K线数据的列表 all_ohlcv = [] # 初始化当前查询的开始时间 current_since = since # 添加请求计数和最大限制 request_count = 0 max_requests = 50 # 最大请求次数，防止无限循环 # 分页加载数据 while request_count < max_requests: request_count += 1 try: # 获取当前页的数据 ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=current_since, limit=batch_size) # 如果没有获取到数据，结束循环 if not ohlcv or len(ohlcv) == 0: break # 将获取到的数据添加到总列表中 all_ohlcv.extend(ohlcv) # 获取最后一条数据的时间戳 last_timestamp = ohlcv[-1][0] # 如果已达到结束时间，结束循环 if until and last_timestamp >= until: break # 如果获取的数据条数小于限制数，说明已经获取完所有数据 if len(ohlcv) < batch_size: break # 更新下一页的开始时间（加1毫秒避免重复） current_since = last_timestamp + 1 except Exception as e: # 如果单个批次失败，继续尝试下一个批次 if current_since: # 尝试增加时间跳过可能的问题时间点 current_since += 60000 # 跳过1分钟 else: break # 防止API请求过于频繁 time.sleep(0.3) # 减少到0.3秒提高效率 # 数据为空的情况 if not all_ohlcv or len(all_ohlcv) == 0: return None # 转换为DataFrame df = pd.DataFrame(all_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']) df['date'] = pd.to_datetime(df['timestamp'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('Asia/Shanghai') # 在客户端进行结束时间过滤 if until: df = df[df['timestamp'] <= until] # 去除重复数据 df = df.drop_duplicates(subset=['timestamp']) # 按时间排序 df = df.sort_values('timestamp') # 限制数据条数的逻辑 - 优先考虑时间范围 if start_time and end_time: # 如果指定了明确的时间范围，返回该时间范围内的所有数据 if len(df) > 10000: # 防止数据量过大，设置一个合理的上限 df = df.tail(10000).reset_index(drop=True) elif limit and len(df) > limit: # 如果没有指定明确时间范围，使用默认的limit限制 df = df.tail(limit).reset_index(drop=True) # 如果过滤后没有数据，返回None if len(df) == 0: return None return df except Exception as e: return None def get_a_stock_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None): """获取A股K线数据""" try: # 处理时间戳参数转换为日期字符串 start_date = None end_date = None if start_time: try: # 尝试解析时间戳（毫秒） start_timestamp = int(start_time) start_date = datetime.fromtimestamp(start_timestamp / 1000).strftime('%Y-%m-%d') except (ValueError, TypeError): # 如果不是时间戳，尝试解析datetime-local格式 (YYYY-MM-DDTHH:MM) try: if 'T' in str(start_time): # datetime-local格式：2025-05-19T06:07 start_date = str(start_time).split('T')[0] # 只取日期部分 else: start_date = str(start_time) except: start_date = start_time if end_time: try: # 尝试解析时间戳（毫秒） end_timestamp = int(end_time) end_date = datetime.fromtimestamp(end_timestamp / 1000).strftime('%Y-%m-%d') except (ValueError, TypeError): # 如果不是时间戳，尝试解析datetime-local格式 try: if 'T' in str(end_time): # datetime-local格式：2025-05-26T06:07 end_date = str(end_time).split('T')[0] # 只取日期部分 else: end_date = str(end_time) except: end_date = end_time # 如果用户指定了时间范围，优先获取该范围内的所有数据 actual_limit = limit if start_date and end_date: actual_limit = None # 不限制数据条数，获取完整时间范围数据 # 调用A股数据获取器 df = china_stock.get_kl_data(symbol, timeframe, start_date, end_date, actual_limit) if df is None: return None return df except Exception as e: return None def add_indicators(df): fast = 8 slow = 16 period = 6 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] df['ma5'] = (ta.MA(df, timeperiod=5)).fillna(0) df['ma10'] = (ta.MA(df, timeperiod=10)).fillna(0) df['ma30'] = (ta.EMA(df, timeperiod=30)).fillna(0) df['ma250'] = (ta.MA(df, timeperiod=250)).fillna(0) df['rsi'] = ta.RSI(df, timeperiod=14) # 计算布林带 (当前周期 - 20周期，2标准差) bb = ta.BBANDS(df, timeperiod=365, nbdevup=3.0, nbdevdn=3.0, matype=0) df['bb_upper'] = bb['upperband'].fillna(0) df['bb_middle'] = bb['middleband'].fillna(0) df['bb_lower'] = bb['lowerband'].fillna(0) # 计算次周期布林带 (14周期，2标准差) bb_element = ta.BBANDS(df, timeperiod=14, nbdevup=2.0, nbdevdn=2.0, matype=0) df['element_bb_upper'] = bb_element['upperband'].fillna(0) df['element_bb_middle'] = bb_element['middleband'].fillna(0) df['element_bb_lower'] = bb_element['lowerband'].fillna(0) df['macd'] = df['macd'].fillna(0) df['macdsignal'] = df['macdsignal'].fillna(0) df['macdhist'] = df['macdhist'].fillna(0) df['ma5'] = df['ma5'].fillna(0) df['ma10'] = df['ma10'].fillna(0) df['ma30'] = df['ma30'].fillna(0) df['ma250'] = df['ma250'].fillna(0) df['rsi'] = df['rsi'].fillna(0) df['avg_volume'] = df['volume'].rolling(10).mean() # 计算量比，避免产生Infinity值 df['volume_ratio'] = df['volume'] / df['avg_volume'] # 填充缺失值（前N根K线） df['volume_ratio'] = df['volume_ratio'].fillna(1.0) df['avg_volume'] = df['avg_volume'].fillna(0) # 处理Infinity和-Infinity值 df['volume_ratio'] = df['volume_ratio'].replace([float('inf'), float('-inf')], 1.0) # 计算ATR (Average True Range) - 14周期 df['atr'] = ta.ATR(df, timeperiod=14) df['atr'] = df['atr'].fillna(0) return df def calculate_macd(df): """计算MACD指标""" exp1 = df['close'].ewm(span=10, adjust=False).mean() exp2 = df['close'].ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal return { 'macd': macd.tolist(), 'signal': signal.tolist(), 'histogram': histogram.tolist() } def analyze_chan(df): """进行缠论分析""" chan = ChanLun() # 获取分析结果 klc_list = chan.get_klc_list(df) bi_list = chan.cal_bi_list(klc_list) #for index in range(0, 10): #print(bi_list[index].start_time, bi_list[index].start_klc.end_time, bi_list[index].dir) seg_list = chan.get_seg_list(bi_list) zs_list = chan.calculate_zs(bi_list, seg_list) # 添加买卖点识别 buy_sell_points = identify_trade_points(bi_list, seg_list, zs_list) for bi in bi_list: bi.cal_macdhist() for bi in bi_list: bi.cal_macd_div() #print(bi.start_time, bi.macd_hist, bi.macd_div) # 获取原始K线数据用于KLU分型分析 klu_list = [] try: # 尝试获取KLU数据 if hasattr(chan, 'get_klu_list'): klu_list = chan.get_klu_list(df) elif hasattr(chan, 'klu_list'): klu_list = chan.klu_list else: # 如果没有专门的KLU方法，尝试从KLC获取原始K线数据 pass except Exception as e: klu_list = [] # 提取K线分型信息 klc_fx_info = [] for klc in klc_list: if hasattr(klc, 'klc_fx_type') and klc.klc_fx_type != Chan_KLC_FX.UNKNOWN: try: # 计算分型强度 fx_strength = 0 fx_strength_level = "" is_strong_fx = False # 统一使用cal_fx_strength函数 if hasattr(klc, 'cal_fx_strength'): fx_strength = klc.cal_fx_strength(5) # 尝试获取分型强度等级 if hasattr(klc, 'get_fx_strength_level'): fx_strength_level = klc.get_fx_strength_level() # 尝试判断是否为强分型 if hasattr(klc, 'is_strong_fx'): is_strong_fx = klc.is_strong_fx() # 如果分型强度小于1，设为0 if fx_strength < 1: fx_strength = 0 klc_fx_info.append({ 'time': klc.end_time, 'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high, 'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), 'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM, 'fx_strength': fx_strength, # 分型强度分数 (0-100) 'fx_strength_level': fx_strength_level, # 分型强度等级 (极强/强/中等/弱/极弱) 'is_strong_fx': is_strong_fx # 是否为强分型 }) except Exception as e: # 如果出错，仍然添加基本信息，但分型强度为0 klc_fx_info.append({ 'time': klc.end_time, 'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high, 'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""), 'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM, 'fx_strength': 0, 'fx_strength_level': "", 'is_strong_fx': False }) # 提取KLU分型信息 klu_fx_info = [] for klu in klu_list: if hasattr(klu, 'fx_type') and klu.fx_type != Chan_FX_TYPE.UNKNOWN: try: # 计算分型强度 fx_strength = 0 fx_strength_level = "" is_strong_fx = False # 尝试调用分型强度计算方法 if hasattr(klu, 'calculate_realtime_fx_strength'): fx_strength = klu.calculate_realtime_fx_strength() elif hasattr(klu, 'fx_strength'): fx_strength = klu.fx_strength # 尝试获取分型强度等级 - 基于强度值生成等级 if fx_strength >= 2: fx_strength_level = "强" is_strong_fx = True elif fx_strength >= 1: fx_strength_level = "中" is_strong_fx = False elif fx_strength >= 0: fx_strength_level = "弱" is_strong_fx = False else: fx_strength_level = "极弱" is_strong_fx = False # 确保分型确认状态 is_confirmed = getattr(klu, 'fx_confirmed', True) klu_fx_info.append({ 'time': klu.time, 'price': klu.low if klu.fx_type == Chan_FX_TYPE.BOTTOM else klu.high, 'fx_type': str(klu.fx_type).replace("Chan_FX_TYPE.", ""), 'is_bottom': klu.fx_type == Chan_FX_TYPE.BOTTOM, 'fx_strength': fx_strength, # 分型强度分数 'fx_strength_level': fx_strength_level, # 分型强度等级 'is_strong_fx': is_strong_fx, # 是否为强分型 'fx_confirmed': is_confirmed # 分型是否确认 }) except Exception as e: # 如果出错，仍然添加基本信息，但分型强度为0 klu_fx_info.append({ 'time': klu.time, 'price': klu.low if klu.fx_type == Chan_FX_TYPE.BOTTOM else klu.high, 'fx_type': str(klu.fx_type).replace("Chan_FX_TYPE.", ""), 'is_bottom': klu.fx_type == Chan_FX_TYPE.BOTTOM, 'fx_strength': 0, 'fx_strength_level': "", 'is_strong_fx': False, 'fx_confirmed': False }) return { 'klc_list': klc_list, 'klu_list': klu_list, # 添加KLU列表 'bi_list': bi_list, 'seg_list': seg_list, 'zs_list': zs_list, 'trade_points': buy_sell_points, 'klc_fx_info': klc_fx_info, # KLC分型信息 'klu_fx_info': klu_fx_info # 添加KLU分型信息 } def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, start_time=None, end_time=None): """生成逐步计算的回放数据""" replay_data = {} # 预先获取完整的次周期数据（避免重复数据获取） element_full_data = None if element_timeframe and symbol: # 一次性获取完整的次周期数据 element_full_data = get_kl_data(symbol, element_timeframe, start_time=start_time, end_time=end_time) if element_full_data is not None and len(element_full_data) > 0: # 一次性添加技术指标 element_full_data = add_indicators(element_full_data) # 为每个K线索引计算分析结果 for i in range(1, len(df) + 1): # 从1开始，至少需要1根K线 try: # 截取到当前索引的数据 current_df = df.iloc[:i].copy() # 添加技术指标 current_df = add_indicators(current_df) # 进行缠论分析 analysis_result = analyze_chan(current_df) # 计算MACD macd_data = calculate_macd(current_df) # 如果有次周期数据，筛选对应时间范围的数据 element_step_data = {} if element_full_data is not None: # 获取当前主周期时间范围 current_end_time = current_df['timestamp'].iloc[-1] if len(current_df) > 0 else None if current_end_time: # 筛选次周期数据：只取时间戳小于等于当前主周期结束时间的数据 element_current_df = element_full_data[element_full_data['timestamp'] <= current_end_time].copy() if len(element_current_df) > 0: # 重新对当前时间范围的次周期数据进行缠论分析 # 这样可以确保数据的准确性，避免时间筛选的复杂性 element_current_analysis = analyze_chan(element_current_df) # 直接使用分析结果，无需复杂的时间筛选 filtered_bi_list = element_current_analysis['bi_list'] filtered_seg_list = element_current_analysis['seg_list'] filtered_zs_list = element_current_analysis['zs_list'] filtered_trade_points = element_current_analysis['trade_points'] filtered_klc_fx = element_current_analysis['klc_fx_info'] filtered_klu_fx = element_current_analysis['klu_fx_info'] # 计算当前时间范围的MACD element_macd_data = calculate_macd(element_current_df) element_step_data = { 'element_kline_data': clean_dataframe_for_json(element_current_df).to_dict('records'), 'element_bi_list': [{ 'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None, 'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None, 'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high, 'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None, 'direction': convert_direction(bi.dir), 'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0 } for bi in filtered_bi_list if bi.end_klc], 'element_seg_list': [{ 'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None, 'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in filtered_seg_list if seg.end_bi], 'element_zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure } for zs in filtered_zs_list if zs.end_klc], 'element_uncompleted_zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure } for zs in filtered_zs_list if not zs.is_sure], 'element_trade_points': [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in filtered_trade_points], 'element_macd': element_macd_data, 'element_bollinger': { 'upper': element_current_df['bb_upper'].tolist(), 'middle': element_current_df['bb_middle'].tolist(), 'lower': element_current_df['bb_lower'].tolist() }, 'element_element_bollinger': { 'upper': element_current_df['element_bb_upper'].tolist(), 'middle': element_current_df['element_bb_middle'].tolist(), 'lower': element_current_df['element_bb_lower'].tolist() }, # 添加次周期ATR数据 'element_atr': element_current_df['atr'].tolist(), 'element_klc_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), 'fx_strength_level': str(point['fx_strength_level']), 'is_strong_fx': bool(point['is_strong_fx']) } for point in filtered_klc_fx], 'element_klu_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), 'fx_strength_level': str(point['fx_strength_level']), 'is_strong_fx': bool(point['is_strong_fx']), 'fx_confirmed': bool(point['fx_confirmed']) } for point in filtered_klu_fx] } # 构建该索引对应的分析结果 step_data = { 'step_index': i-1, # 当前步骤索引 'total_steps': len(df), # 总步骤数 'has_element_data': element_timeframe is not None and len(element_step_data) > 0, # 是否包含次周期数据 'element_timeframe': element_timeframe, # 次周期时间框架 'kline_data': clean_dataframe_for_json(current_df).to_dict('records'), 'bi_list': [{ 'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None, 'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None, 'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high, 'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None, 'direction': convert_direction(bi.dir), 'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0 } for bi in analysis_result['bi_list'] if bi.end_klc], 'seg_list': [{ 'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None, 'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in analysis_result['seg_list'] if seg.end_bi], 'zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure } for zs in analysis_result['zs_list'] if zs.end_klc], 'uncompleted_zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure } for zs in analysis_result['zs_list'] if not zs.is_sure], 'trade_points': [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in analysis_result['trade_points']], 'macd': macd_data, 'bollinger': { 'upper': current_df['bb_upper'].tolist(), 'middle': current_df['bb_middle'].tolist(), 'lower': current_df['bb_lower'].tolist() }, 'element_bollinger': { 'upper': current_df['element_bb_upper'].tolist(), 'middle': current_df['element_bb_middle'].tolist(), 'lower': current_df['element_bb_lower'].tolist() }, # 添加ATR数据 'atr': current_df['atr'].tolist(), 'klc_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']) # 是否为强分型 } for point in analysis_result['klc_fx_info']], 'klu_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型 'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认 } for point in analysis_result['klu_fx_info']] } # 合并次周期数据到step_data中，如果没有次周期数据则提供空的占位符 if element_step_data: step_data.update(element_step_data) else: # 提供空的次周期数据结构，确保前端可以统一处理 step_data.update({ 'element_kline_data': [], 'element_bi_list': [], 'element_seg_list': [], 'element_zs_list': [], 'element_uncompleted_zs_list': [], 'element_trade_points': [], 'element_macd': {'macd': [], 'signal': [], 'histogram': []}, 'element_bollinger': {'upper': [], 'middle': [], 'lower': []}, 'element_element_bollinger': {'upper': [], 'middle': [], 'lower': []}, 'element_atr': [], 'element_klc_fx_info': [], 'element_klu_fx_info': [] }) replay_data[i-1] = step_data # 使用0-based索引 except Exception as e: continue return replay_data def identify_trade_points(bi_list, seg_list, zs_list): """识别缠论买卖点 - 多级别识别，减少滞后性""" trade_points = [] # 1. 基于笔的二三类买卖点识别（更及时） trade_points.extend(identify_bi_trade_points(bi_list, zs_list)) # 2. 基于线段的一类买卖点识别（传统方法） trade_points.extend(identify_seg_trade_points(seg_list)) # 3. 基于分型强度的预警点识别（最及时） trade_points.extend(identify_fx_warning_points(bi_list)) # 4. 基于MACD背驰的买卖点识别 trade_points.extend(identify_macd_divergence_points(bi_list)) # 按时间排序 trade_points.sort(key=lambda x: x['time']) return trade_points def identify_bi_trade_points(bi_list, zs_list): """基于笔识别二三类买卖点 - 更及时的信号""" trade_points = [] if len(bi_list) < 3: return trade_points # 构建中枢映射，便于快速查找 zs_map = {} for zs in zs_list: if zs.is_sure: # 只考虑已确认的中枢 zs_map[zs.start_klc.end_time] = zs for i in range(2, len(bi_list)): current_bi = bi_list[i] prev_bi = bi_list[i-1] prev_prev_bi = bi_list[i-2] # 确保笔已完成 if not current_bi.end_klc or not prev_bi.end_klc or not prev_prev_bi.end_klc: continue # 二类买点：向下笔后的向上笔，且不创新低 if (convert_direction(prev_bi.dir) == -1 and convert_direction(current_bi.dir) == 1): prev_low = prev_bi.end_klc.low current_end_price = current_bi.end_klc.high # 检查是否不创新低（相对于前面的低点） if i >= 4: # 至少需要5个笔来判断 earlier_lows = [bi.end_klc.low for bi in bi_list[max(0, i-4):i-1] if convert_direction(bi.dir) == -1 and bi.end_klc] if earlier_lows and prev_low > min(earlier_lows): trade_points.append({ 'type': TRADE_POINT_TYPE.BUY2, 'time': current_bi.end_klc.end_time, 'price': current_end_price, 'desc': '二类买点(笔)' }) # 二类卖点：向上笔后的向下笔，且不创新高 if (convert_direction(prev_bi.dir) == 1 and convert_direction(current_bi.dir) == -1): prev_high = prev_bi.end_klc.high current_end_price = current_bi.end_klc.low # 检查是否不创新高（相对于前面的高点） if i >= 4: # 至少需要5个笔来判断 earlier_highs = [bi.end_klc.high for bi in bi_list[max(0, i-4):i-1] if convert_direction(bi.dir) == 1 and bi.end_klc] if earlier_highs and prev_high < max(earlier_highs): trade_points.append({ 'type': TRADE_POINT_TYPE.SELL2, 'time': current_bi.end_klc.end_time, 'price': current_end_price, 'desc': '二类卖点(笔)' }) return trade_points def identify_seg_trade_points(seg_list): """基于线段识别一类买卖点 - 传统方法""" trade_points = [] if len(seg_list) >= 3: for i in range(2, len(seg_list)): # 确保线段已完成 if seg_list[i].end_bi and seg_list[i-1].end_bi and seg_list[i-2].end_bi: # 一类买点：向下-向上-向下的底分型 if (convert_direction(seg_list[i-2].dir) == -1 and convert_direction(seg_list[i-1].dir) == 1 and convert_direction(seg_list[i].dir) == -1): trade_points.append({ 'type': TRADE_POINT_TYPE.BUY1, 'time': seg_list[i].end_bi.end_klc.end_time, 'price': seg_list[i].end_bi.end_klc.low, 'desc': '一类买点(线段)' }) # 一类卖点：向上-向下-向上的顶分型 if (convert_direction(seg_list[i-2].dir) == 1 and convert_direction(seg_list[i-1].dir) == -1 and convert_direction(seg_list[i].dir) == 1): trade_points.append({ 'type': TRADE_POINT_TYPE.SELL1, 'time': seg_list[i].end_bi.end_klc.end_time, 'price': seg_list[i].end_bi.end_klc.high, 'desc': '一类卖点(线段)' }) return trade_points def identify_fx_warning_points(bi_list): """基于分型强度识别预警点 - 最及时的信号""" trade_points = [] if len(bi_list) < 2: return trade_points # 检查最近的几个笔 recent_bis = bi_list[-3:] if len(bi_list) >= 3 else bi_list for bi in recent_bis: if not bi.end_klc: continue # 获取分型强度（如果有的话） fx_strength = 0 if hasattr(bi.end_klc, 'cal_fx_strength'): try: fx_strength = bi.end_klc.cal_fx_strength(5) except: fx_strength = 0 # 强分型预警（分型强度>=2） if fx_strength >= 2: if convert_direction(bi.dir) == -1: # 向下笔结束，可能的底部 trade_points.append({ 'type': TRADE_POINT_TYPE.BUY3, 'time': bi.end_klc.end_time, 'price': bi.end_klc.low, 'desc': f'强分型预警-买点(强度:{fx_strength})' }) elif convert_direction(bi.dir) == 1: # 向上笔结束，可能的顶部 trade_points.append({ 'type': TRADE_POINT_TYPE.SELL3, 'time': bi.end_klc.end_time, 'price': bi.end_klc.high, 'desc': f'强分型预警-卖点(强度:{fx_strength})' }) return trade_points def identify_macd_divergence_points(bi_list): """基于MACD背驰识别买卖点""" trade_points = [] if len(bi_list) < 4: return trade_points # 检查最近的笔是否有背驰 for i in range(2, len(bi_list)): current_bi = bi_list[i] if not current_bi.end_klc or not hasattr(current_bi, 'macd_div'): continue # MACD背驰阈值 divergence_threshold = 0.3 # 向下笔的底背驰 -> 买点 if (convert_direction(current_bi.dir) == -1 and hasattr(current_bi, 'macd_div') and current_bi.macd_div > divergence_threshold): trade_points.append({ 'type': TRADE_POINT_TYPE.BUY2, 'time': current_bi.end_klc.end_time, 'price': current_bi.end_klc.low, 'desc': f'MACD底背驰买点(背驰度:{current_bi.macd_div:.2f})' }) # 向上笔的顶背驰 -> 卖点 elif (convert_direction(current_bi.dir) == 1 and hasattr(current_bi, 'macd_div') and current_bi.macd_div > divergence_threshold): trade_points.append({ 'type': TRADE_POINT_TYPE.SELL2, 'time': current_bi.end_klc.end_time, 'price': current_bi.end_klc.high, 'desc': f'MACD顶背驰卖点(背驰度:{current_bi.macd_div:.2f})' }) return trade_points # 辅助函数，转换缠论方向枚举为整数 def convert_direction(direction): """转换方向枚举为数字""" if direction == Chan_BI_DIR.UP or direction == Chan_SEG_DIR.UP: return 1 elif direction == Chan_BI_DIR.DOWN or direction == Chan_SEG_DIR.DOWN: return -1 else: return 0 def format_time_safely(time_obj, client_tz): """安全地格式化时间对象，处理字符串和datetime两种情况""" if time_obj is None: return None if isinstance(time_obj, str): # 尝试将字符串解析为datetime try: from dateutil import parser time_obj = parser.parse(time_obj) return time_obj.astimezone(client_tz).isoformat() except: return time_obj else: # 已经是datetime对象 return time_obj.astimezone(client_tz).isoformat() def is_smaller_timeframe(tf1, tf2): """判断时间周期tf1是否小于tf2""" # 定义时间周期的分钟数映射 tf_values = { '1m': 1, '3m': 3, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '2h': 120, '4h': 240, '6h': 360, '8h': 480, '12h': 720, '1d': 1440, '3d': 4320, '1w': 10080, '1M': 43200 } # 获取时间周期对应的分钟数 tf1_value = tf_values.get(tf1) tf2_value = tf_values.get(tf2) # 如果某个时间周期不在映射中，返回False if tf1_value is None or tf2_value is None: return False # 返回tf1是否小于tf2 return tf1_value < tf2_value def is_smaller_or_equal_timeframe(tf1, tf2): """判断时间周期tf1是否小于等于tf2""" # 定义时间周期的分钟数映射 tf_values = { '1m': 1, '3m': 3, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '2h': 120, '4h': 240, '6h': 360, '8h': 480, '12h': 720, '1d': 1440, '3d': 4320, '1w': 10080, '1M': 43200 } # 获取时间周期对应的分钟数 tf1_value = tf_values.get(tf1) tf2_value = tf_values.get(tf2) # 如果某个时间周期不在映射中，返回False if tf1_value is None or tf2_value is None: return False # 返回tf1是否小于等于tf2 return tf1_value <= tf2_value def clean_dataframe_for_json(df): """清理DataFrame数据用于JSON序列化""" # 创建副本避免修改原始数据 clean_df = df.copy() # 替换NaN值为None clean_df = clean_df.where(pd.notnull(clean_df), None) return clean_df @app.route('/') def index(): """主页""" return render_template('index.html', timeframes=TIMEFRAMES, symbols=SYMBOLS, a_stock_symbols=A_STOCK_SYMBOLS) @app.route('/api/analyze') def analyze(): """分析接口""" symbol = request.args.get('symbol', 'SOL/USDT:USDT') timeframe = request.args.get('timeframe', '5m') # 验证交易对不为空 if not symbol or symbol.strip() == '': return jsonify({'error': '交易对不能为空'}) # 获取时间范围参数 start_time = request.args.get('start_time') end_time = request.args.get('end_time') # 获取客户端请求的时区 client_timezone = request.args.get('timezone', 'Asia/Shanghai') # 获取分形元素时间周期 element_timeframe = request.args.get('element_timeframe') # 获取是否只需要分形元素数据的参数 elements_only_param = request.args.get('elements_only') elements_only = elements_only_param == 'true' # 获取是否需要回放数据的参数 need_replay_data = request.args.get('need_replay_data', 'false').lower() == 'true' # 验证小周期是否小于主周期 if element_timeframe and not is_smaller_or_equal_timeframe(element_timeframe, timeframe): return jsonify({'error': '分形元素时间周期必须小于或等于主图表时间周期'}) # 获取数据 df = get_kl_data(symbol, timeframe, start_time=start_time, end_time=end_time) if df is None: return jsonify({'error': '获取数据失败'}) if len(df) == 0: return jsonify({'error': '所选时间范围内没有数据'}) # 使用客户端指定的时区 client_tz = timezone(client_timezone) # 如果只需要分形元素数据而不需要主周期数据，则初始化一个空结果 result = { 'timezone': client_timezone } # 如果不是只需要分形元素数据，则添加主周期数据 if not elements_only: # 添加技术指标（包括布林带） df = add_indicators(df) # 进行缠论分析 analysis_result = analyze_chan(df) # 计算MACD macd_data = calculate_macd(df) # 如果需要回放数据，生成逐步计算的回放数据 if need_replay_data: replay_data = generate_replay_data(df, client_tz, symbol, element_timeframe, start_time, end_time) else: replay_data = None # 添加主周期分析结果到返回数据 result.update({ 'kline_data': clean_dataframe_for_json(df).to_dict('records'), 'bi_list': [{ 'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None, 'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None, 'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high, 'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None, 'direction': convert_direction(bi.dir), 'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0 } for bi in analysis_result['bi_list'] if bi.end_klc], 'seg_list': [{ 'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None, 'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in analysis_result['seg_list'] if seg.end_bi], 'zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 添加中枢是否完成的标志 } for zs in analysis_result['zs_list'] if zs.end_klc], # 添加未完成中枢列表 'uncompleted_zs_list': [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': None, # 未完成中枢没有结束时间 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in analysis_result['zs_list'] if not zs.is_sure], 'trade_points': [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in analysis_result['trade_points']], 'macd': macd_data, # 添加布林带数据 'bollinger': { 'upper': df['bb_upper'].tolist(), 'middle': df['bb_middle'].tolist(), 'lower': df['bb_lower'].tolist() }, 'element_bollinger': { 'upper': df['element_bb_upper'].tolist(), 'middle': df['element_bb_middle'].tolist(), 'lower': df['element_bb_lower'].tolist() }, # 添加ATR数据 'atr': df['atr'].tolist(), # 添加K线分型信息 'klc_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']) # 是否为强分型 } for point in analysis_result['klc_fx_info']], 'klu_fx_info': [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型 'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认 } for point in analysis_result['klu_fx_info']] }) # 如果生成了回放数据，添加到返回结果中 if replay_data is not None: result['replay_data'] = replay_data # 如果有指定分形元素时间周期，获取小周期数据 if element_timeframe: # 获取小周期数据，使用与主周期相同的时间范围 element_df = get_kl_data(symbol, element_timeframe, start_time=start_time, end_time=end_time) if element_df is not None and len(element_df) > 0: # 添加小周期技术指标（包括布林带） element_df = add_indicators(element_df) # 对小周期数据进行缠论分析 element_analysis = analyze_chan(element_df) # 计算小周期MACD数据 element_macd_data = calculate_macd(element_df) # 添加小周期分析结果到返回数据 result['element_timeframe'] = element_timeframe result['element_macd'] = element_macd_data # 添加小周期MACD数据 # 添加小周期布林带数据 result['element_bollinger'] = { 'upper': element_df['bb_upper'].tolist(), 'middle': element_df['bb_middle'].tolist(), 'lower': element_df['bb_lower'].tolist() } result['element_element_bollinger'] = { 'upper': element_df['element_bb_upper'].tolist(), 'middle': element_df['element_bb_middle'].tolist(), 'lower': element_df['element_bb_lower'].tolist() } # 添加小周期ATR数据 result['element_atr'] = element_df['atr'].tolist() result['element_bi_list'] = [{ 'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None, 'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None, 'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high, 'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None, 'direction': convert_direction(bi.dir), 'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0 } for bi in element_analysis['bi_list'] if bi.end_klc] # 添加小周期K线数据 result['element_kline_data'] = clean_dataframe_for_json(element_df).to_dict('records') result['element_seg_list'] = [{ 'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None, 'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in element_analysis['seg_list'] if seg.end_bi] result['element_zs_list'] = [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 添加中枢是否完成的标志 } for zs in element_analysis['zs_list'] if zs.end_klc] result['element_uncompleted_zs_list'] = [{ 'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(), 'end_time': None, # 未完成中枢没有结束时间 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in element_analysis['zs_list'] if not zs.is_sure] result['element_trade_points'] = [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in element_analysis['trade_points']] # 添加小周期分型信息 result['element_klc_fx_info'] = [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']) # 是否为强分型 } for point in element_analysis['klc_fx_info']] result['element_klu_fx_info'] = [{ 'time': format_time_safely(point['time'], client_tz), 'price': float(point['price']), 'fx_type': point['fx_type'], 'is_bottom': bool(point['is_bottom']), 'fx_strength': float(point['fx_strength']), # 分型强度分数 'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级 'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型 'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认 } for point in element_analysis['klu_fx_info']] pass return jsonify(result) @app.route('/api/symbols') def get_symbols(): """获取可用交易对""" try: markets = exchange.load_markets() # 合约交易对通常是以USDT结尾的永续合约 symbols = [symbol for symbol in markets.keys() if '/USDT' in symbol and ':USDT' in symbol] return jsonify(symbols) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/a_stocks') def get_a_stocks(): """获取A股股票列表""" try: stock_list = china_stock.get_stock_list() return jsonify(stock_list) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/popular_a_stocks') def get_popular_a_stocks(): """获取热门A股股票""" try: return jsonify(china_stock.get_popular_stocks()) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/sectors') def get_sectors(): """获取所有行业分类""" try: sectors = china_stock.get_all_sectors() return jsonify(sectors) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/stocks_by_sector') def get_stocks_by_sector(): """根据行业获取股票""" try: sector = request.args.get('sector') if sector: stocks = china_stock.get_stock_by_sector(sector) return jsonify(stocks) else: # 返回所有行业的股票分组 all_sectors = china_stock.get_stock_by_sector() return jsonify(all_sectors) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/search_stock') def search_stock(): """搜索股票 - 增强版""" try: keyword = request.args.get('keyword', '') if not keyword: return jsonify({'error': '搜索关键词不能为空'}) results = china_stock.search_stock(keyword) return jsonify(results) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/test_element_data') def test_element_data(): """测试次周期数据是否正确生成""" try: symbol = request.args.get('symbol', 'SOL/USDT:USDT') timeframe = request.args.get('timeframe', '1h') element_timeframe = request.args.get('element_timeframe', '15m') # 获取主周期数据 main_df = get_kl_data(symbol, timeframe, limit=3) if main_df is None or len(main_df) == 0: return jsonify({'error': '无法获取主周期数据'}) # 获取次周期数据 element_df = get_kl_data(symbol, element_timeframe, start_time=main_df['timestamp'].iloc[0], end_time=main_df['timestamp'].iloc[-1]) if element_df is None or len(element_df) == 0: return jsonify({'error': '无法获取次周期数据'}) # 分析次周期数据 element_df = add_indicators(element_df) element_analysis = analyze_chan(element_df) return jsonify({ 'main_data_count': len(main_df), 'element_data_count': len(element_df), 'element_analysis': { 'bi_count': len(element_analysis['bi_list']), 'seg_count': len(element_analysis['seg_list']), 'zs_count': len(element_analysis['zs_list']), 'klc_fx_count': len(element_analysis['klc_fx_info']), 'klu_fx_count': len(element_analysis['klu_fx_info']), 'trade_points_count': len(element_analysis['trade_points']) }, 'sample_bi': [{'has_end_klc': bi.end_klc is not None, 'direction': convert_direction(bi.dir)} for bi in element_analysis['bi_list'][:2]] if len(element_analysis['bi_list']) > 0 else [], 'sample_klc_fx': element_analysis['klc_fx_info'][:3] if len(element_analysis['klc_fx_info']) > 0 else [], 'sample_klu_fx': element_analysis['klu_fx_info'][:3] if len(element_analysis['klu_fx_info']) > 0 else [] }) except Exception as e: import traceback return jsonify({'error': str(e), 'traceback': traceback.format_exc()}) @app.route('/api/debug_replay_sample') def debug_replay_sample(): """调试接口：返回回放数据样本，方便前端调试""" try: symbol = request.args.get('symbol', 'SOL/USDT:USDT') timeframe = request.args.get('timeframe', '1h') element_timeframe = request.args.get('element_timeframe', '15m') step = int(request.args.get('step', 2)) # 返回第几步的数据 # 获取少量数据进行测试 df = get_kl_data(symbol, timeframe, limit=5) if df is None or len(df) == 0: return jsonify({'error': '无法获取测试数据'}) # 生成回放数据 client_tz = timezone('Asia/Shanghai') replay_data = generate_replay_data( df, client_tz, symbol, element_timeframe, start_time=None, end_time=None ) if step not in replay_data: return jsonify({'error': f'步骤 {step} 不存在，可用步骤：{list(replay_data.keys())}'}) # 返回指定步骤的完整数据 step_data = replay_data[step] return jsonify({ 'step': step, 'data': step_data, 'summary': { 'has_element_data': step_data.get('has_element_data', False), 'element_timeframe': step_data.get('element_timeframe'), 'main_bi_count': len(step_data.get('bi_list', [])), 'main_klc_fx_count': len(step_data.get('klc_fx_info', [])), 'main_klu_fx_count': len(step_data.get('klu_fx_info', [])), 'element_bi_count': len(step_data.get('element_bi_list', [])), 'element_klc_fx_count': len(step_data.get('element_klc_fx_info', [])), 'element_klu_fx_count': len(step_data.get('element_klu_fx_info', [])), 'element_kline_count': len(step_data.get('element_kline_data', [])) } }) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/debug_replay_structure') def debug_replay_structure(): """调试接口：检查回放数据结构""" try: # 获取一个简单的测试案例 symbol = request.args.get('symbol', 'SOL/USDT:USDT') timeframe = request.args.get('timeframe', '1h') element_timeframe = request.args.get('element_timeframe', '15m') # 获取少量数据进行测试 df = get_kl_data(symbol, timeframe, limit=5) # 只取5根K线 if df is None or len(df) == 0: return jsonify({'error': '无法获取测试数据'}) # 生成回放数据 client_tz = timezone('Asia/Shanghai') replay_data = generate_replay_data( df, client_tz, symbol, element_timeframe, start_time=None, end_time=None ) # 返回结构信息 result = { 'total_steps': len(replay_data), 'sample_step_keys': list(replay_data[0].keys()) if len(replay_data) > 0 else [], 'has_element_data_in_steps': [], 'element_data_counts': {} } # 检查每个步骤的次周期数据 for step_idx, step_data in replay_data.items(): has_element = step_data.get('has_element_data', False) result['has_element_data_in_steps'].append({ 'step': step_idx, 'has_element_data': has_element, 'element_bi_count': len(step_data.get('element_bi_list', [])), 'element_klc_fx_count': len(step_data.get('element_klc_fx_info', [])), 'element_klu_fx_count': len(step_data.get('element_klu_fx_info', [])) }) return jsonify(result) except Exception as e: return jsonify({'error': str(e)}) @app.route('/api/filter_stocks', methods=['POST']) def filter_stocks(): """筛选满足条件的A股股票""" try: data = request.get_json() start_time = data.get('start_time') end_time = data.get('end_time') timeframe = data.get('timeframe', '1d') fx_strength_threshold = data.get('fx_strength_threshold', 1.0) if not start_time or not end_time: return jsonify({'error': '开始时间和结束时间不能为空'}) # 获取所有A股股票列表，如果失败则使用热门股票作为备用 stock_list = [] data_source = "" try: stock_list = china_stock.get_stock_list() if stock_list and len(stock_list) > 0: data_source = "完整股票列表" else: raise Exception("获取到的股票列表为空") except Exception as e: try: popular_stocks = china_stock.get_popular_stocks() stock_list = [{'symbol': stock['symbol'], 'name': stock['name']} for stock in popular_stocks] data_source = "热门股票列表" except Exception as e2: # 检查是否是网络连接问题 if "timeout" in str(e).lower() or "connection" in str(e).lower() or "network" in str(e).lower(): return jsonify({ 'error': '网络连接超时，无法获取股票数据。请检查网络连接后重试。', 'error_type': 'network_error', 'suggestion': '请确保网络连接正常，或稍后重试。' }) else: return jsonify({'error': f'无法获取股票列表: {str(e)}'}) if not stock_list: return jsonify({ 'error': '无法获取股票列表，请检查网络连接后重试', 'error_type': 'network_error', 'suggestion': '请确保网络连接正常，或稍后重试。' }) results = [] processed_count = 0 total_count = len(stock_list) failed_count = 0 for stock in stock_list: try: symbol = stock['symbol'] name = stock['name'] processed_count += 1 # 获取股票K线数据 df = get_a_stock_kl_data(symbol, timeframe, start_time=start_time, end_time=end_time) if df is None or len(df) < 3: failed_count += 1 # 如果连续失败太多，可能是网络问题 if failed_count > 10 and len(results) == 0: return jsonify({ 'error': '网络连接不稳定，无法获取股票数据。请检查网络连接后重试。', 'error_type': 'network_error', 'processed_count': processed_count, 'failed_count': failed_count }) continue # 进行缠论分析 analysis_result = analyze_chan(df) if not analysis_result or 'klc_fx_info' not in analysis_result: continue klc_fx_info = analysis_result['klc_fx_info'] # 检查最近2个KLC是否有满足条件的分型 recent_klcs = klc_fx_info[-2:] if len(klc_fx_info) >= 2 else klc_fx_info for klc_info in recent_klcs: fx_strength = klc_info.get('fx_strength', 0) fx_type = klc_info.get('fx_type', 'UNKNOWN') # 检查是否满足条件：分型强度>=阈值 且 分型类型不为UNKNOWN if fx_strength >= fx_strength_threshold and fx_type != 'UNKNOWN': # 获取当前价格（最新收盘价） current_price = df['close'].iloc[-1] if len(df) > 0 else None fx_price = klc_info.get('price', 0) # 计算涨跌幅 change_percent = 0 if current_price and fx_price and fx_price > 0: change_percent = ((current_price - fx_price) / fx_price) * 100 # 格式化分型类型显示 fx_type_display = format_fx_type(fx_type) results.append({ 'symbol': symbol, 'name': name, 'fx_time': klc_info.get('time', ''), 'fx_type': fx_type_display, 'fx_strength': fx_strength, 'fx_price': fx_price, 'current_price': current_price, 'change_percent': change_percent }) break # 找到一个满足条件的就跳出循环 except Exception as e: failed_count += 1 continue # 按分型强度降序排列 results.sort(key=lambda x: x['fx_strength'], reverse=True) return jsonify({ 'results': results, 'total_processed': processed_count, 'total_found': len(results), 'failed_count': failed_count, 'data_source': data_source, 'message': f'使用{data_source}进行筛选，共处理{processed_count}只股票，找到{len(results)}只满足条件的股票' }) except Exception as e: # 检查是否是网络连接问题 if "timeout" in str(e).lower() or "connection" in str(e).lower() or "network" in str(e).lower(): return jsonify({ 'error': '网络连接超时，请检查网络连接后重试。', 'error_type': 'network_error', 'suggestion': '请确保网络连接正常，或稍后重试。' }) else: return jsonify({'error': str(e)}) def format_fx_type(fx_type): """格式化分型类型显示""" fx_type_map = { 'TOP1': '顶分型1', 'TOP2': '顶分型2', 'TOP3': '顶分型3', 'BOTTOM1': '底分型1', 'BOTTOM2': '底分型2', 'BOTTOM3': '底分型3', 'TOP': '顶分型', 'BOTTOM': '底分型' } return fx_type_map.get(fx_type, fx_type) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=8128)