from __future__ import annotations import sys import os from collections import OrderedDict import json import logging import time import traceback import io import base64 from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta import ccxt import numpy as np import pandas as pd import requests import talib.abstract as ta from pytz import timezone _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _ROOT not in sys.path: sys.path.append(_ROOT) from chanlun import ChanLun, TF_DF from chanlun.core.ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR from chanlun.indicators.ChanMACD import ChanMACD from chanlun.analysis.ChanZone import StructureZoneConfig, analyze_structure_zones_from_serialized from config import ( DATA_SERVICE_URL, MACD_FAST, MACD_SLOW, MACD_SIGNAL, ccxt_proxies, ) from services.cn_stock import ChinaStockData logger = logging.getLogger(__name__) class TRADE_POINT_TYPE: BUY1 = 1 # 一类买点 BUY2 = 2 # 二类买点 BUY3 = 3 # 三类买点 SELL1 = -1 # 一类卖点 SELL2 = -2 # 二类卖点 SELL3 = -3 # 三类卖点 # mutable runtime state macd_fast_period = MACD_FAST macd_slow_period = MACD_SLOW macd_signal_period = MACD_SIGNAL _proxies = ccxt_proxies() _exchange_kwargs = {"enableRateLimit": True} if _proxies: _exchange_kwargs["proxies"] = _proxies exchange = ccxt.binance(_exchange_kwargs) china_stock = ChinaStockData() _zone_cache = {} 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 _zone_cache_ttl(tf_name: str) -> int: """根据时间周期返回缓存过期时间（秒）""" minutes = timeframe_to_minutes(tf_name) or 5 if minutes <= 5: return 120 # 5m及以下: 2分钟 elif minutes <= 15: return 300 # 15m: 5分钟 elif minutes <= 60: return 600 # 1h: 10分钟 else: return 1800 # 4h+: 30分钟 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 compute_timeframe_defaults(labels_ordered): """ 根据已排序的「周期 → 中文标签」映射，计算主 / 次 / 次次周期默认值。 labels_ordered: OrderedDict 或按插入顺序排列的 dict。 """ if not labels_ordered: labels_ordered = DEFAULT_TIMEFRAME_LABELS.copy() timeframe_keys = list(labels_ordered.keys()) preferred_main = next((tf for tf in ['5m', '15m', '1h'] if tf in labels_ordered), None) default_main = preferred_main or (timeframe_keys[0] if timeframe_keys else '1m') if default_main not in labels_ordered 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 if timeframe_keys: try: idx_el = timeframe_keys.index(default_element) default_sub_sub = timeframe_keys[idx_el - 1] if idx_el > 0 else timeframe_keys[0] except ValueError: default_sub_sub = timeframe_keys[0] else: default_sub_sub = default_element return default_main, default_element, default_sub_sub, timeframe_keys 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) if limit is not None: params["limit"] = limit 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 股下拉仅放默认一项；用户切换到「A股」时由前端请求 /api/a_stocks 填充全市场（约 5500+） A_STOCK_SYMBOLS = [{'symbol': '000001', 'name': '平安银行'}] 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): global macd_fast_period, macd_slow_period, macd_signal_period macd = ta.MACD(df, fastperiod=macd_fast_period, slowperiod=macd_slow_period, signalperiod=macd_signal_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) df['ema26'] = (ta.EMA(df, timeperiod=26)).fillna(0) df['ema13'] = (ta.EMA(df, timeperiod=13)).fillna(0) df['ema7'] = (ta.EMA(df, timeperiod=7)).fillna(0) df['ema104'] = (ta.EMA(df, timeperiod=104)).fillna(0) df['ema156'] = (ta.EMA(df, timeperiod=156)).fillna(0) df['ema208'] = (ta.EMA(df, timeperiod=208)).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) bb2633 = ta.BBANDS(df, timeperiod=26, nbdevup=3.0, nbdevdn=3.0, matype=0) bbp2633 = (df['close'] - bb2633['lowerband']) / (bb2633['upperband'] - bb2633['lowerband']) df['bb2633upper'] = bb2633['upperband'].fillna(0) df['bb2633lower'] = bb2633['lowerband'].fillna(0) df['bbp2633'] = bbp2633.fillna(0) df['bb2633middle'] = bb2633['middleband'].fillna(0) return df def calculate_macd(df): """计算MACD指标""" global macd_fast_period, macd_slow_period, macd_signal_period exp1 = df['close'].ewm(span=macd_fast_period, adjust=False).mean() exp2 = df['close'].ewm(span=macd_slow_period, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=macd_signal_period, 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 = TF_DF() # 初始化多时间周期数据以获取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_seg_zs(seg_list) # 计算笔中枢（BI中枢）并拍平成列表 #bi_zs_list = chan.cal_bi_zs_list_pure(bi_list) bi_zs_list = chan.cal_bi_zs(seg_list) bsp_list = [] if len(bi_zs_list) > 0: bsp_list = chan.find_all_bsp(bi_list, bi_zs_list) #bsp_state_list = chan.get_bsp_state(df) #for bsp in bsp_list: #print(bsp.end_time, bsp.type, bsp.dir) # 添加买卖点识别 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分析（复用 get_klc_list 内已算好的结果，避免同周期二次全量分析） chan_macd = None chan_macd_data = {} try: if klu_list and len(klu_list) > 0: print(f"获取到KLU列表，长度: {len(klu_list)}") chan_macd = getattr(chan, '_last_chan_macd', None) if chan_macd is None: 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 # KLC 分型框（起止时间+高低价）： # 仅使用 cal_fx_box 通过 display 条件后生成的 klc.fx_box。 # 若无 fx_box，则前端不应绘制分型框。 fx_box = getattr(klc, 'fx_box', None) box_start_time = getattr(fx_box, 'start_time', None) if fx_box else None box_end_time = getattr(fx_box, 'end_time', None) if fx_box else None box_high = getattr(fx_box, 'high', None) if fx_box else None box_low = getattr(fx_box, 'low', None) if fx_box else None 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, # 是否为强分型 # 虚线分型框信息（给前端画框用） 'start_time': box_start_time, 'end_time': box_end_time, 'high': float(box_high) if box_high is not None else None, 'low': float(box_low) if box_low is not None else None, }) except Exception as e: # 如果出错，仍然添加基本信息，但分型强度为0 fx_box = getattr(klc, 'fx_box', None) box_start_time = getattr(fx_box, 'start_time', None) if fx_box else None box_end_time = getattr(fx_box, 'end_time', None) if fx_box else None box_high = getattr(fx_box, 'high', None) if fx_box else None box_low = getattr(fx_box, 'low', None) if fx_box else None 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, # 虚线分型框信息（给前端画框用） 'start_time': box_start_time, 'end_time': box_end_time, 'high': float(box_high) if box_high is not None else None, 'low': float(box_low) if box_low is not None else None, }) 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中枢列表 'bsp_list': bsp_list, # 添加买卖点列表 '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] 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