from flask import Flask, render_template, jsonify, request from collections import OrderedDict import json import logging import ccxt import pandas as pd import requests from datetime import datetime, timedelta import sys import os 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, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR from cn_stock_data import ChinaStockData from ChanMACD import ChanMACD # 添加买卖点枚举类型 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() logger = logging.getLogger(__name__) DATA_SERVICE_URL = os.environ.get("DATA_SERVICE_URL", os.environ.get("DATASVC_URL", "http://127.0.0.1:9009")) #DATA_SERVICE_URL = os.environ.get("DATA_SERVICE_URL", os.environ.get("DATASVC_URL", "http://192.168.1.9:9009")) DEFAULT_TIMEFRAME_LABELS = OrderedDict([ ("1m", "1分钟"), ("3m", "3分钟"), ("5m", "5分钟"), ("15m", "15分钟"), ("30m", "30分钟"), ("1h", "1小时"), ("2h", "2小时"), ("4h", "4小时"), ("6h", "6小时"), ("8h", "8小时"), ("12h", "12小时"), ("1d", "日线"), ("3d", "3日线"), ("1w", "周线"), ("1M", "月线"), ]) DEFAULT_SYMBOLS = [ 'SOL/USDT:USDT', 'BTC/USDT:USDT', 'ETH/USDT:USDT', 'BNB/USDT:USDT', 'XRP/USDT:USDT', 'WIF/USDT:USDT', 'ADA/USDT:USDT', 'DOGE/USDT:USDT', 'AVAX/USDT:USDT', 'DOT/USDT:USDT', 'MATIC/USDT:USDT' ] TIMEFRAMES = DEFAULT_TIMEFRAME_LABELS.copy() SYMBOLS = DEFAULT_SYMBOLS.copy() DATA_SERVICE_AVAILABLE = False SERVICE_METADATA_LAST_REFRESH = 0 def timeframe_to_minutes(tf: str): """将时间周期转换为分钟数，用于排序。""" if not tf: return None unit = tf[-1] try: value = int(tf[:-1]) except (ValueError, TypeError): return None multiplier = { 'm': 1, 'h': 60, 'd': 1440, 'w': 10080, 'M': 43200, # 30天近似 }.get(unit) if multiplier is None: return None return value * multiplier def format_timeframe_label(tf: str) -> str: """将时间周期转换为可读标签。""" if not tf: return tf unit = tf[-1] try: value = int(tf[:-1]) except (ValueError, TypeError): return tf if unit == 'm': return f"{value}分钟" if unit == 'h': return f"{value}小时" if unit == 'd': return "日线" if value == 1 else f"{value}日线" if unit == 'w': return "周线" if value == 1 else f"{value}周线" if unit == 'M': return "月线" if value == 1 else f"{value}月线" return tf def build_timeframe_labels(timeframes): ordered = sorted( timeframes, key=lambda tf: timeframe_to_minutes(tf) if timeframe_to_minutes(tf) is not None else float('inf'), ) labels = OrderedDict() for tf in ordered: labels[tf] = format_timeframe_label(tf) return labels def _parse_time_input(value): if value in (None, '', 0): return None try: return int(float(value)) except (ValueError, TypeError): return None def refresh_data_service_metadata(force=False): """刷新数据服务提供的交易对与周期元信息。""" global DATA_SERVICE_AVAILABLE, TIMEFRAMES, SYMBOLS, SERVICE_METADATA_LAST_REFRESH now = time.time() if not force and DATA_SERVICE_AVAILABLE and now - SERVICE_METADATA_LAST_REFRESH < 60: return True try: resp = requests.get(f"{DATA_SERVICE_URL}/health", timeout=5) resp.raise_for_status() payload = resp.json() service_symbols = payload.get("symbols") or payload.get("symbol_list") or [] base_timeframes = payload.get("timeframes") or payload.get("base_timeframes") or [] derived = payload.get("derived_timeframes") or [] service_timeframes = list(base_timeframes) for tf in derived: if tf not in service_timeframes: service_timeframes.append(tf) if service_symbols: SYMBOLS[:] = service_symbols if service_timeframes: TIMEFRAMES.clear() TIMEFRAMES.update(build_timeframe_labels(service_timeframes)) DATA_SERVICE_AVAILABLE = True SERVICE_METADATA_LAST_REFRESH = now return True except Exception as exc: logger.warning("无法加载数据服务元信息: %s", exc) if not DATA_SERVICE_AVAILABLE: TIMEFRAMES.clear() TIMEFRAMES.update(DEFAULT_TIMEFRAME_LABELS) SYMBOLS[:] = DEFAULT_SYMBOLS DATA_SERVICE_AVAILABLE = False return False def _fetch_kl_from_datasvc(symbol, timeframe, start_ms=None, end_ms=None, limit=None): params = {"symbol": symbol, "tf": timeframe} if start_ms is not None: params["start"] = int(start_ms) if end_ms is not None: params["end"] = int(end_ms) resp = requests.get(f"{DATA_SERVICE_URL}/api/candles", params=params, timeout=10) resp.raise_for_status() data = resp.json() if not data: return None df = pd.DataFrame(data) if df.empty or "timestamp" not in df.columns: return None numeric_cols = ["open", "high", "low", "close", "volume"] df["timestamp"] = pd.to_numeric(df["timestamp"], errors="coerce") df = df.dropna(subset=["timestamp"]) df["timestamp"] = df["timestamp"].astype("int64") for col in numeric_cols: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.dropna(subset=numeric_cols) df = df.sort_values("timestamp") if limit and len(df) > limit: df = df.tail(limit) df = df.reset_index(drop=True) df["date"] = pd.to_datetime(df["timestamp"], unit='ms', utc=True).dt.tz_convert('Asia/Shanghai') return df # 模块加载时尝试预取一次元信息，但失败不阻塞后续流程 refresh_data_service_metadata(force=True) # 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=100000, 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_via_ccxt(symbol, timeframe, limit=100000, 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 = 1000 # 分钟级数据减少批次大小 elif timeframe in ['15m', '30m', '1h']: batch_size = 1000 else: batch_size = 1500 # 日线及以上可以获取更多 batch_size = 1500 # 默认批次大小 # 初始化存储所有K线数据的列表 all_ohlcv = [] # 初始化当前查询的开始时间 current_since = since # 添加请求计数和最大限制 request_count = 0 max_requests = 300 # 最大请求次数，防止无限循环 # 分页加载数据 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) > 100000: # 防止数据量过大，设置一个合理的上限 df = df.tail(100000).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_crypto_kl_data(symbol, timeframe, limit=100000, start_time=None, end_time=None): """优先通过本地数据服务获取加密货币K线，失败时回退至交易所API。""" start_ms = _parse_time_input(start_time) end_ms = _parse_time_input(end_time) refresh_data_service_metadata() if DATA_SERVICE_AVAILABLE: try: df = _fetch_kl_from_datasvc( symbol=symbol, timeframe=timeframe, start_ms=start_ms, end_ms=end_ms, limit=limit, ) if df is not None and not df.empty: return df except Exception as exc: logger.warning("数据服务请求失败，准备回退至交易所 API：%s", exc) return _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit, start_time, end_time) def get_a_stock_kl_data(symbol, timeframe, limit=100000, 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 = 12*1 slow = 26*1 period = 9*1 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) # 新增 EMA 指标 df['ema5'] = (ta.EMA(df, timeperiod=5)).fillna(0) df['ema10'] = (ta.EMA(df, timeperiod=10)).fillna(0) df['ema24'] = (ta.EMA(df, timeperiod=24)).fillna(0) df['ema52'] = (ta.EMA(df, timeperiod=52)).fillna(0) # 常用SMA 24/52 try: df['sma24'] = (ta.SMA(df, timeperiod=24)).fillna(0) df['sma52'] = (ta.SMA(df, timeperiod=52)).fillna(0) except Exception: df['sma24'] = 0 df['sma52'] = 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) bb30 = ta.BBANDS(df, timeperiod=41, nbdevup=2.3, nbdevdn=2.3, matype=0) #bb30 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) df['bbup30'] = bb30['upperband'].fillna(0) df['bblow30'] = bb30['lowerband'].fillna(0) bb302 = ta.BBANDS(df, timeperiod=41, nbdevup=2.0, nbdevdn=2.0, matype=0) #bb302 = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) df['bbup302'] = bb302['upperband'].fillna(0) df['bblow302'] = bb302['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['ema5'] = df['ema5'].fillna(0) df['ema10'] = df['ema10'].fillna(0) df['ema24'] = df['ema24'].fillna(0) df['ema52'] = df['ema52'].fillna(0) df['sma24'] = df['sma24'].fillna(0) df['sma52'] = df['sma52'].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=12, 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, symbol=None, timeframe=None): """进行缠论分析""" chan = ChanLun() # 初始化多时间周期数据以获取EMA52 ema52_dict = None # 获取分析结果 klu_list = chan.get_kl_data(df) klc_list = chan.get_klc_list(klu_list) 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) # 计算笔中枢（BI中枢）并拍平成列表 try: bi_zs_nested = chan.cal_bi_zs(seg_list) bi_zs_list = [zs for group in bi_zs_nested for zs in (group or [])] if bi_zs_nested else [] except Exception: bi_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) # 添加ChanMACD分析 chan_macd = None chan_macd_data = {} try: if klu_list and len(klu_list) > 0: print(f"获取到KLU列表，长度: {len(klu_list)}") chan_macd = ChanMACD(klu_list) chan_macd_data = { 'seg_list': chan_macd.seg_list, 'unittf_list': chan_macd.unittf_list, 'histset_list': chan_macd.histset_list, 'klu_list': chan_macd.klu_list, 'high_position_list': chan_macd.high_position_list, 'high_empty_list': chan_macd.high_empty_list, 'low_position_list': getattr(chan_macd, 'low_position_list', []), 'low_empty_list': getattr(chan_macd, 'low_empty_list', []), 'return_zero_list': chan_macd.return_zero_list, 'cross0_up_list': chan_macd.cross0_up_list, 'cross0_down_list': chan_macd.cross0_down_list } print(f"ChanMACD分析完成: seg={len(chan_macd.seg_list)}, unittf={len(chan_macd.unittf_list)}, histset={len(chan_macd.histset_list)}") else: print("未能获取KLU列表或列表为空") chan_macd_data = { 'seg_list': [], 'unittf_list': [], 'histset_list': [], 'high_position_list': [], 'high_empty_list': [], 'return_zero_list': [], 'cross0_up_list': [], 'cross0_down_list': [] } except Exception as e: print(f"ChanMACD分析出错: {e}") import traceback traceback.print_exc() chan_macd_data = { 'seg_list': [], 'unittf_list': [], 'histset_list': [], 'high_position_list': [], 'high_empty_list': [], 'low_position_list': [], 'low_empty_list': [], 'return_zero_list': [], 'cross0_up_list': [], 'cross0_down_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 if klc.bb_out: 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 }) return { 'klc_list': klc_list, 'klu_list': klu_list, # 添加KLU列表 'bi_list': bi_list, 'seg_list': seg_list, 'zs_list': zs_list, 'bi_zs_list': bi_zs_list, # 添加BI中枢列表 'klc_fx_info': klc_fx_info, # KLC分型信息 'chan_macd': chan_macd_data, # 添加ChanMACD分析数据 'ema52_dict': ema52_dict # 添加多时间周期EMA52数据 } # 辅助函数，转换缠论方向枚举为整数 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 serialize_chan_macd_data(chan_macd_data, client_tz): """序列化ChanMACD数据为JSON可序列化格式""" serialized_data = { 'seg_list': [], 'unittf_list': [], 'histset_list': [], # 状态标记数据 'high_position_list': [], 'high_empty_list': [], 'low_position_list': [], 'low_empty_list': [], 'return_zero_list': [], 'cross0_up_list': [], 'cross0_down_list': [], # 新增：输出KLU的继续背驰/分离背驰标志 'klu_list': [] } # 序列化seg_list for seg in chan_macd_data.get('seg_list', []): try: seg_data = { 'start_time': format_time_safely(seg.start_time, client_tz), 'end_time': format_time_safely(seg.end_time, client_tz) if seg.end_time else None, 'seg_dir': 'ABOVE' if seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else 'UNDER', 'klu_count': len(seg.klu_list) if hasattr(seg, 'klu_list') else 0, 'unittf_count': len(seg.unittf_list) if hasattr(seg, 'unittf_list') else 0, 'histset_count': len(seg.hist_set) if hasattr(seg, 'hist_set') else 0 } serialized_data['seg_list'].append(seg_data) except Exception as e: print(f"序列化seg出错: {e}") continue # 序列化unittf_list（兼容新结构与枚举类型） for unittf in chan_macd_data.get('unittf_list', []): try: dir_value = getattr(unittf, 'uinttf_dir', None) dir_name = getattr(dir_value, 'name', dir_value if isinstance(dir_value, str) else None) start_t = getattr(unittf, 'start_type', None) start_type = getattr(start_t, 'name', start_t) end_t = getattr(unittf, 'end_type', None) end_type = getattr(end_t, 'name', end_t) peak_abs = getattr(unittf, 'peak_abs', None) if peak_abs is None: peak_abs = getattr(unittf, 'peak_hist', None) length = getattr(unittf, 'length', None) if length is None: length = len(unittf.klu_list) if hasattr(unittf, 'klu_list') else None unittf_data = { 'start_time': format_time_safely(getattr(unittf, 'start_time', None), client_tz), 'end_time': format_time_safely(getattr(unittf, 'end_time', None), client_tz) if getattr(unittf, 'end_time', None) else None, 'dir': dir_name, # 'ABOVE' | 'UNDER' | None 'start_type': start_type, # e.g. 'START' | 'CROSS0' | 'NEAR0_UP' | 'NEAR0_DOWN' 'end_type': end_type, 'invalid': getattr(unittf, 'invalid', False), 'peak_abs': peak_abs, 'length': length, 'klu_count': len(unittf.klu_list) if hasattr(unittf, 'klu_list') else 0, 'histset_count': len(unittf.histset_list) if hasattr(unittf, 'histset_list') else 0 } serialized_data['unittf_list'].append(unittf_data) except Exception as e: print(f"序列化unittf出错: {e}") continue # 序列化histset_list for histset in chan_macd_data.get('histset_list', []): try: histset_data = { 'start_time': format_time_safely(getattr(histset, 'start_time', None), client_tz), 'end_time': format_time_safely(getattr(histset, 'end_time', None), client_tz), 'histset_dir': 'ABOVE' if histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE else 'UNDER', 'klu_count': len(histset.klu_list) if hasattr(histset, 'klu_list') else 0 } serialized_data['histset_list'].append(histset_data) except Exception as e: print(f"序列化histset出错: {e}") continue # 序列化状态标记数据 # 序列化高位列表 for high_pos in chan_macd_data.get('high_position_list', []): try: high_pos_data = { 'time': format_time_safely(high_pos['time'], client_tz), 'end_time': format_time_safely(high_pos.get('end_time'), client_tz) if high_pos.get('end_time') else None, 'type': high_pos.get('type', 'start'), 'macd': high_pos.get('macd'), 'signal': high_pos.get('signal'), 'macdhist': high_pos.get('macdhist'), 'end_macd': high_pos.get('end_macd'), 'end_signal': high_pos.get('end_signal'), 'end_macdhist': high_pos.get('end_macdhist') } serialized_data['high_position_list'].append(high_pos_data) except Exception as e: print(f"序列化high_position出错: {e}") continue # 序列化高位空列表 for high_empty in chan_macd_data.get('high_empty_list', []): try: high_empty_data = { 'time': format_time_safely(high_empty['time'], client_tz), 'end_time': format_time_safely(high_empty.get('end_time'), client_tz) if high_empty.get('end_time') else None, 'type': high_empty.get('type', 'start'), 'macd': high_empty.get('macd'), 'signal': high_empty.get('signal'), 'macdhist': high_empty.get('macdhist'), 'end_macd': high_empty.get('end_macd'), 'end_signal': high_empty.get('end_signal'), 'end_macdhist': high_empty.get('end_macdhist') } serialized_data['high_empty_list'].append(high_empty_data) except Exception as e: print(f"序列化high_empty出错: {e}") continue # 序列化低位与低位空 for low_pos in chan_macd_data.get('low_position_list', []): try: low_pos_data = { 'time': format_time_safely(low_pos['time'], client_tz), 'end_time': format_time_safely(low_pos.get('end_time'), client_tz) if low_pos.get('end_time') else None, 'type': low_pos.get('type', 'start'), 'macd': low_pos.get('macd'), 'signal': low_pos.get('signal'), 'macdhist': low_pos.get('macdhist'), 'end_macd': low_pos.get('end_macd'), 'end_signal': low_pos.get('end_signal'), 'end_macdhist': low_pos.get('end_macdhist') } serialized_data['low_position_list'].append(low_pos_data) except Exception as e: print(f"序列化low_position出错: {e}") continue for low_empty in chan_macd_data.get('low_empty_list', []): try: low_empty_data = { 'time': format_time_safely(low_empty['time'], client_tz), 'end_time': format_time_safely(low_empty.get('end_time'), client_tz) if low_empty.get('end_time') else None, 'type': low_empty.get('type', 'start'), 'macd': low_empty.get('macd'), 'signal': low_empty.get('signal'), 'macdhist': low_empty.get('macdhist'), 'end_macd': low_empty.get('end_macd'), 'end_signal': low_empty.get('end_signal'), 'end_macdhist': low_empty.get('end_macdhist') } serialized_data['low_empty_list'].append(low_empty_data) except Exception as e: print(f"序列化low_empty出错: {e}") continue # 序列化归零轴列表 for return_zero in chan_macd_data.get('return_zero_list', []): try: return_zero_data = { 'time': format_time_safely(return_zero['time'], client_tz), 'end_time': format_time_safely(return_zero.get('end_time'), client_tz) if return_zero.get('end_time') else None, 'type': return_zero.get('type', 'start'), 'macd': return_zero.get('macd'), 'signal': return_zero.get('signal'), 'macdhist': return_zero.get('macdhist'), 'end_macd': return_zero.get('end_macd'), 'end_signal': return_zero.get('end_signal'), 'end_macdhist': return_zero.get('end_macdhist') } serialized_data['return_zero_list'].append(return_zero_data) except Exception as e: print(f"序列化return_zero出错: {e}") continue # 序列化穿越零轴列表 for cross0_up in chan_macd_data.get('cross0_up_list', []): try: cross0_up_data = { 'time': format_time_safely(cross0_up['time'], client_tz), 'type': cross0_up.get('type', 'start'), 'macd': cross0_up.get('macd'), 'signal': cross0_up.get('signal'), 'macdhist': cross0_up.get('macdhist') } serialized_data['cross0_up_list'].append(cross0_up_data) except Exception as e: print(f"序列化cross0_up出错: {e}") continue for cross0_down in chan_macd_data.get('cross0_down_list', []): try: cross0_down_data = { 'time': format_time_safely(cross0_down['time'], client_tz), 'type': cross0_down.get('type', 'start'), 'macd': cross0_down.get('macd'), 'signal': cross0_down.get('signal'), 'macdhist': cross0_down.get('macdhist') } serialized_data['cross0_down_list'].append(cross0_down_data) except Exception as e: print(f"序列化cross0_down出错: {e}") continue # 序列化 KLU 列表（仅导出需要的时间与背驰标志） for klu in chan_macd_data.get('klu_list', []): try: serialized_data['klu_list'].append({ 'time': format_time_safely(getattr(klu, 'time', None), client_tz), 'continue_div': bool(getattr(klu, 'continue_div', False)), 'separate_div': int(getattr(klu, 'separate_div', 0)) if getattr(klu, 'separate_div', 0) is not None else 0, 'near0_return': int(getattr(klu, 'near0_return', 0)) if getattr(klu, 'near0_return', 0) is not None else 0 }) except Exception as e: print(f"序列化klu出错: {e}") continue return serialized_data def is_smaller_timeframe(tf1, tf2): """判断时间周期tf1是否小于tf2""" tf1_value = timeframe_to_minutes(tf1) tf2_value = timeframe_to_minutes(tf2) if tf1_value is None or tf2_value is None: return False return tf1_value < tf2_value def is_smaller_or_equal_timeframe(tf1, tf2): """判断时间周期tf1是否小于等于tf2""" tf1_value = timeframe_to_minutes(tf1) tf2_value = timeframe_to_minutes(tf2) if tf1_value is None or tf2_value is None: return False 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 # ====== 趋势判定与趋势筛选（币对） ====== def classify_trend_stage(df): """根据 EMA 斜率与多空排列判断趋势方向与阶段 返回: direction in {"bull","bear","sideways"}, stage in {"early","mid","late"}, strength_score (0-100) """ if df is None or len(df) < 60: return "sideways", "early", 0 # 使用 EMA5/10/24/52 closes = df['close'].values ema5 = df['ema5'].values if 'ema5' in df else ta.EMA(df, timeperiod=5) ema10 = df['ema10'].values if 'ema10' in df else ta.EMA(df, timeperiod=10) ema24 = df['ema24'].values if 'ema24' in df else ta.EMA(df, timeperiod=24) ema52 = df['ema52'].values if 'ema52' in df else ta.EMA(df, timeperiod=52) # 最近N根用于斜率与排列判定 lookback = min(30, len(df) - 1) if lookback <= 5: return "sideways", "early", 0 # 简单斜率: 最近k根的线性变化率近似 def slope(arr, k=10): k = min(k, len(arr) - 1) if k < 2: return 0.0 y = arr[-k:] x = np.arange(k) # 最小二乘拟合斜率 denom = np.dot(x - x.mean(), x - x.mean()) if denom == 0: return 0.0 m = np.dot(y - y.mean(), x - x.mean()) / denom return float(m) k_slope = 12 # 斜率窗口 s5 = slope(ema5, k_slope) s10 = slope(ema10, k_slope) s24 = slope(ema24, k_slope) s52 = slope(ema52, k_slope) # 多空排列 last5, last10, last24, last52 = ema5[-1], ema10[-1], ema24[-1], ema52[-1] bull_stack = last5 > last10 > last24 > last52 bear_stack = last5 < last10 < last24 < last52 # 波动性与动量增强: MACD 柱体最近均值 macdhist = df['macdhist'].values if 'macdhist' in df else calculate_macd(df)['histogram'] hist_recent = macdhist[-lookback:] hist_power = float(np.mean(np.abs(hist_recent))) if len(hist_recent) else 0.0 # 方向 if bull_stack and s24 > 0 and s52 > 0: direction = "bull" elif bear_stack and s24 < 0 and s52 < 0: direction = "bear" else: # 用价格相对 EMA52 辅助 if closes[-1] > last52 and (s24 + s52) > 0: direction = "bull" elif closes[-1] < last52 and (s24 + s52) < 0: direction = "bear" else: direction = "sideways" # 阶段: 依据（斜率大小、与EMA52距离、MACD柱体扩张/收敛） dist52 = float((closes[-1] - last52) / last52) if last52 else 0.0 slope_score = max(0.0, (abs(s24) + abs(s52)) * 1000.0) # 归一化 dist_score = min(50.0, abs(dist52) * 200.0) hist_score = min(30.0, hist_power * 10.0) strength = float(min(100.0, slope_score + dist_score + hist_score)) # 简单阶段判定 if direction == "sideways": stage = "early" strength = min(strength, 30.0) else: # 查看最近 hist 是否在扩大或收敛 if len(hist_recent) >= 6: recent_growth = np.mean(np.abs(hist_recent[-3:])) - np.mean(np.abs(hist_recent[-6:-3])) else: recent_growth = 0.0 if recent_growth > 0 and abs(dist52) < 0.05: stage = "early" elif recent_growth > 0 and abs(dist52) >= 0.05: stage = "mid" else: stage = "late" return direction, stage, strength def load_crypto_symbols(limit=200): """加载常见USDT永续合约交易对，返回列表""" refresh_data_service_metadata() if SYMBOLS: return SYMBOLS[:limit] try: markets = exchange.load_markets() symbols = [s for s in markets.keys() if '/USDT' in s and ':USDT' in s] return symbols[:limit] except Exception: return DEFAULT_SYMBOLS[:limit] @app.route('/api/trend_filter', methods=['GET']) def trend_filter(): """趋势筛选接口（币对） 参数: timeframe: K线周期 start_time, end_time: 毫秒时间戳，可选 direction: bull/bear/sideways 可选 stage: early/mid/late 可选 min_strength: 0-100 可选 symbols: 逗号分隔列表，可选；不传则自动加载部分USDT币对 返回符合条件的币对与简要统计 """ timeframe = request.args.get('timeframe', '1h') start_time = request.args.get('start_time') end_time = request.args.get('end_time') want_direction = request.args.get('direction') # 可为 None want_stage = request.args.get('stage') # 可为 None try: min_strength = float(request.args.get('min_strength', '0')) except ValueError: min_strength = 0.0 symbols_param = request.args.get('symbols') if symbols_param: symbols_list = [s.strip() for s in symbols_param.split(',') if s.strip()] else: symbols_list = load_crypto_symbols(limit=150) results = [] for sym in symbols_list: try: df = get_crypto_kl_data(sym, timeframe, start_time=start_time, end_time=end_time) if df is None or len(df) < 60: continue df = add_indicators(df) direction, stage, strength = classify_trend_stage(df) if want_direction and direction != want_direction: continue if want_stage and stage != want_stage: continue if strength < min_strength: continue last_row = df.iloc[-1] results.append({ 'symbol': sym, 'time': int(last_row['timestamp']), 'close': float(last_row['close']), 'direction': direction, 'stage': stage, 'strength': float(round(strength, 2)), 'ema5': float(last_row['ema5']), 'ema10': float(last_row['ema10']), 'ema24': float(last_row['ema24']), 'ema52': float(last_row['ema52']) }) except Exception: continue # 按强度降序 results.sort(key=lambda x: x['strength'], reverse=True) return jsonify({ 'count': len(results), 'results': results }) @app.route('/api/trend_detail', methods=['GET']) def trend_detail(): """返回单个币对的K线与EMA、用于前端绘制趋势线 参数: symbol, timeframe, start_time, end_time """ symbol = request.args.get('symbol') timeframe = request.args.get('timeframe', '1h') start_time = request.args.get('start_time') end_time = request.args.get('end_time') timezone_name = request.args.get('timezone', 'Asia/Shanghai') if not symbol: return jsonify({'error': 'symbol不能为空'}) df = get_crypto_kl_data(symbol, timeframe, start_time=start_time, end_time=end_time) if df is None or len(df) == 0: return jsonify({'error': '获取数据失败'}) df = add_indicators(df) direction, stage, strength = classify_trend_stage(df) # 简单趋势线: 用最近N根收盘价做线性拟合 N = min(80, len(df)) sub = df.tail(N) y = sub['close'].values x = np.arange(len(y)) denom = np.dot(x - x.mean(), x - x.mean()) if denom != 0: m = float(np.dot(y - y.mean(), x - x.mean()) / denom) b = float(y.mean() - m * x.mean()) else: m, b = 0.0, float(y[-1]) client_tz = timezone(timezone_name) return jsonify({ 'symbol': symbol, 'timeframe': timeframe, 'timezone': timezone_name, 'direction': direction, 'stage': stage, 'strength': float(round(strength, 2)), 'kline_data': clean_dataframe_for_json(df)[['timestamp','open','high','low','close','volume','ema5','ema10','ema24','ema52']].to_dict('records'), 'trend_line': { 'offset': int(df.index[-N]), 'slope': m, 'intercept': b, 'length': int(N) } }) @app.route('/') def index(): """主页""" refresh_data_service_metadata() timeframe_items = list(TIMEFRAMES.items()) timeframe_keys = [item[0] for item in timeframe_items] symbols = SYMBOLS if SYMBOLS else DEFAULT_SYMBOLS preferred_main = next((tf for tf in ['5m', '15m', '1h'] if tf in TIMEFRAMES), None) default_main = preferred_main or (timeframe_keys[0] if timeframe_keys else '1m') if default_main not in TIMEFRAMES and timeframe_keys: default_main = timeframe_keys[0] if timeframe_keys: try: idx = timeframe_keys.index(default_main) default_element = timeframe_keys[idx - 1] if idx > 0 else timeframe_keys[0] except ValueError: default_element = timeframe_keys[0] else: default_element = default_main default_symbol = 'BTC/USDT:USDT' if 'BTC/USDT:USDT' in symbols else (symbols[0] if symbols else '') return render_template( 'index.html', timeframes=TIMEFRAMES, symbols=symbols, a_stock_symbols=A_STOCK_SYMBOLS, default_main_timeframe=default_main, default_element_timeframe=default_element, default_symbol=default_symbol, timeframe_keys_json=json.dumps(timeframe_keys), data_service_available=DATA_SERVICE_AVAILABLE, ) @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' # 验证小周期是否小于主周期 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, symbol, timeframe) # 计算MACD macd_data = calculate_macd(df) # 基于已有 KLC 列表生成趋势标记（不做额外计算） klc_trend = [] try: for klc in analysis_result.get('klc_list', []): trend_val = getattr(klc, 'trend', None) t_obj = getattr(klc, 'end_time', None) or getattr(klc, 'start_time', None) if trend_val is None or t_obj is None: continue # 统一成字符串：UP/DOWN/FLAT/UNKNOWN trend_name = str(trend_val) if '.' in trend_name: trend_name = trend_name.split('.')[-1] time_str = format_time_safely(t_obj, client_tz) if time_str: klc_trend.append({'time': time_str, 'trend': trend_name}) except Exception: klc_trend = [] # 添加主周期分析结果到返回数据 result.update({ 'kline_data': clean_dataframe_for_json(df).to_dict('records'), 'klc_list': [{ 'date': klc.end_time if isinstance(klc.end_time, str) else klc.end_time.astimezone(client_tz).isoformat(), 'open': float(klc.open), 'high': float(klc.high), 'low': float(klc.low), 'close': float(klc.close), 'volume': float(klc.volume) if hasattr(klc, 'volume') else 0, 'direction': str(klc.dir).replace('Chan_KLINE_DIR.', ''), 'fx_type': str(klc.fx).replace('Chan_FX_TYPE.', ''), 'klc_fx_type': str(klc.klc_fx_type).replace('Chan_KLC_FX.', ''), 'trend': str(klc.trend).replace('Chan_PRICE_TREND.', '') } for klc in analysis_result['klc_list'] if hasattr(klc, 'end_time') and klc.end_time], '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], # 添加未完成笔列表 'uncompleted_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': 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': 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 not 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.is_sure], # 添加未完成线段列表 'uncompleted_seg_list': get_uncompleted_seg_list(analysis_result['seg_list'], client_tz), '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, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': zs.is_sure # 添加中枢是否完成的标志 } for zs in analysis_result['zs_list'] if zs.end_klc], # 添加主周期BI中枢列表（已完成） '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()) if getattr(zs.start_klc, 'end_time', None) else (zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat()) ), 'end_time': (zs.end_time if isinstance(zs.end_time, str) else zs.end_time.astimezone(client_tz).isoformat()) if getattr(zs, 'end_time', None) else None, 'zg': zs.zg, 'zd': zs.zd, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': bool(getattr(zs, 'is_sure', False)) } for zs in analysis_result.get('bi_zs_list', []) if getattr(zs, 'is_sure', False)], # 添加未完成中枢列表 '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, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in analysis_result['zs_list'] if not zs.is_sure], # 添加未完成BI中枢列表 'uncompleted_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()) if getattr(zs.start_klc, 'end_time', None) else (zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat()) ), 'end_time': None, 'zg': zs.zg, 'zd': zs.zd, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': bool(getattr(zs, 'is_sure', False)) } for zs in analysis_result.get('bi_zs_list', []) if not getattr(zs, 'is_sure', False)], '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']], # 添加ChanMACD分析数据 'chan_macd': serialize_chan_macd_data(analysis_result.get('chan_macd', {}), client_tz), # 添加多时间周期EMA52数据 'ema52_dict': analysis_result.get('ema52_dict', {}), # 直接输出KLC趋势标记（使用已有trend字段） 'klc_trend': klc_trend }) # 如果有指定分形元素时间周期，获取小周期数据 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, symbol, element_timeframe) # 计算小周期MACD数据 element_macd_data = calculate_macd(element_df) # 组装小周期 KLC 趋势（仅提取已有 trend，不做重算） try: element_klc_trend = [] for klc in element_analysis.get('klc_list', []): trend_val = getattr(klc, 'trend', None) t_obj = getattr(klc, 'end_time', None) or getattr(klc, 'start_time', None) if trend_val is None or t_obj is None: continue trend_name = str(trend_val) if '.' in trend_name: trend_name = trend_name.split('.')[-1] time_str = format_time_safely(t_obj, client_tz) if time_str: element_klc_trend.append({'time': time_str, 'trend': trend_name}) except Exception: element_klc_trend = [] # 添加小周期分析结果到返回数据 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] # 添加次周期未完成笔列表 result['element_uncompleted_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': 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': 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 not bi.end_klc] # 添加小周期K线数据 result['element_kline_data'] = clean_dataframe_for_json(element_df).to_dict('records') # 添加小周期KLC列表 result['element_klc_list'] = [{ 'date': klc.end_time if isinstance(klc.end_time, str) else klc.end_time.astimezone(client_tz).isoformat(), 'open': float(klc.open), 'high': float(klc.high), 'low': float(klc.low), 'close': float(klc.close), 'volume': float(klc.volume) if hasattr(klc, 'volume') else 0, 'direction': str(klc.dir).replace('Chan_KLINE_DIR.', ''), 'fx_type': str(klc.fx).replace('Chan_FX_TYPE.', ''), 'klc_fx_type': str(klc.klc_fx_type).replace('Chan_KLC_FX.', ''), 'trend': str(klc.trend).replace('Chan_PRICE_TREND.', '') } for klc in element_analysis['klc_list'] if hasattr(klc, 'end_time') and klc.end_time] 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.is_sure] # 添加次周期未完成线段列表 result['element_uncompleted_seg_list'] = get_uncompleted_seg_list(element_analysis['seg_list'], client_tz) 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, 'gg': zs.gg, 'dd': zs.dd, '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, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in element_analysis['zs_list'] if not zs.is_sure] # 添加次周期 BI 中枢（已完成/未完成） result['element_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()) if getattr(zs.start_klc, 'end_time', None) else (zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat()) ), 'end_time': (zs.end_time if isinstance(zs.end_time, str) else zs.end_time.astimezone(client_tz).isoformat()) if getattr(zs, 'end_time', None) else None, 'zg': zs.zg, 'zd': zs.zd, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': bool(getattr(zs, 'is_sure', False)) } for zs in element_analysis.get('bi_zs_list', []) if getattr(zs, 'is_sure', False)] result['element_uncompleted_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()) if getattr(zs.start_klc, 'end_time', None) else (zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat()) ), 'end_time': None, 'zg': zs.zg, 'zd': zs.zd, 'gg': zs.gg, 'dd': zs.dd, 'is_sure': bool(getattr(zs, 'is_sure', False)) } for zs in element_analysis.get('bi_zs_list', []) if not getattr(zs, 'is_sure', False)] # 添加小周期分型信息 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']] # 添加次周期ChanMACD分析数据 result['element_chan_macd'] = serialize_chan_macd_data(element_analysis.get('chan_macd', {}), client_tz) # 添加小周期 KLC 趋势标记 result['element_klc_trend'] = element_klc_trend pass return jsonify(result) @app.route('/api/symbols') def get_symbols(): """获取可用交易对""" refresh_data_service_metadata() if SYMBOLS: return jsonify(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(DEFAULT_SYMBOLS) @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)}) def get_uncompleted_seg_list(seg_list, client_tz): """获取未完成线段列表，正确处理倒数第二个和最后一个未完成线段""" uncompleted_segs = [seg for seg in seg_list if not seg.is_sure] if len(uncompleted_segs) == 0: return [] result = [] for i, seg in enumerate(uncompleted_segs): is_last = (i == len(uncompleted_segs) - 1) # 是否为最后一个未完成线段 seg_data = { '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(), '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, 'direction': convert_direction(seg.dir) } if is_last: # 最后一个未完成线段：没有结束时间和价格 seg_data['end_time'] = None seg_data['end_price'] = None else: # 倒数第二个及之前的未完成线段：使用实际的结束时间和价格 if seg.end_bi and seg.end_bi.end_klc: seg_data['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() seg_data['end_price'] = seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low else: # 如果没有结束笔，设为None seg_data['end_time'] = None seg_data['end_price'] = None result.append(seg_data) return result if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=8128)