Files
2025-10-24 23:32:46 +08:00

1421 lines
60 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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__))))
# 导入chan.py项目的核心模块
from Chan import CChan
from ChanConfig import CChanConfig
from Common.CEnum import AUTYPE, DATA_SRC, KL_TYPE, BI_DIR, SEG_DIR, FX_TYPE
from Common.CTime import CTime
from DataAPI.CommonStockAPI import CCommonStockApi
from KLine.KLine_Unit import CKLine_Unit
# 从原有的cn_stock_data导入A股数据获取
try:
from cn_stock_data import ChinaStockData
china_stock = ChinaStockData()
except ImportError:
print("警告: 无法导入cn_stock_dataA股功能将不可用")
china_stock = None
# 添加买卖点枚举类型
class TRADE_POINT_TYPE:
BUY1 = 1 # 一类买点
BUY2 = 2 # 二类买点
BUY3 = 3 # 三类买点
SELL1 = -1 # 一类卖点
SELL2 = -2 # 二类卖点
SELL3 = -3 # 三类卖点
app = Flask(__name__)
# 初始化交易所
exchange = ccxt.binance({
'enableRateLimit': True,
})
# 时间周期映射
TIMEFRAMES = {
'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': '月线',
}
# 常见交易对
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() if china_stock else []
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:
print(f"未知的交易对类型: {symbol}")
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:
print(f"无效的起始时间: {start_time}")
# 结束时间处理
until = None
if end_time:
try:
until = int(end_time)
except ValueError:
print(f"无效的结束时间: {end_time}")
# 根据时间周期调整每次请求的数据量
batch_size = 1000 # 默认批次大小
if timeframe in ['1m', '3m', '5m']:
batch_size = 500 # 分钟级数据减少批次大小
elif timeframe in ['15m', '30m', '1h', '2h']:
batch_size = 1000
elif timeframe in ['4h', '6h', '8h', '12h']:
batch_size = 1200 # 小时级数据可以获取更多
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:
print(f"获取加密货币数据错误: {e}")
traceback.print_exc()
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:
print(f"获取A股数据错误: {e}")
traceback.print_exc()
return None
def add_indicators(df):
fast = 24
slow = 52
period = 18
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)
# 计算ATR (平均真实波幅)
df['atr'] = ta.ATR(df, timeperiod=14).fillna(0)
# 计算布林带 (当前周期 - 20周期,2标准差)
bb = ta.BBANDS(df, timeperiod=20, nbdevup=2.0, nbdevdn=2.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)
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, timeframe='1d'):
"""使用chan.py项目进行缠论分析"""
try:
# 转换时间周期映射 - 根据用户选择的实际时间周期进行映射
timeframe_map = {
'1m': KL_TYPE.K_1M,
'3m': KL_TYPE.K_3M,
'5m': KL_TYPE.K_5M,
'15m': KL_TYPE.K_15M,
'30m': KL_TYPE.K_30M,
'1h': KL_TYPE.K_60M,
'2h': KL_TYPE.K_60M, # 2小时用1小时替代
'4h': KL_TYPE.K_4H,
'6h': KL_TYPE.K_6H,
'8h': KL_TYPE.K_8H,
'12h': KL_TYPE.K_12H,
'1d': KL_TYPE.K_DAY,
'3d': KL_TYPE.K_3DAY,
'1w': KL_TYPE.K_WEEK,
'1M': KL_TYPE.K_MON,
}
# 根据传入的timeframe选择对应的级别
kl_type = timeframe_map.get(timeframe, KL_TYPE.K_DAY)
# 创建配置 - 针对不同时间周期优化买卖点识别
config = CChanConfig({
"bi_strict": True,
"trigger_step": False,
"skip_step": 0,
"divergence_rate": float("inf"),
"bsp2_follow_1": False,
"bsp3_follow_1": False,
"min_zs_cnt": 2,
"bs1_peak": False,
"macd_algo": "peak",
"bs_type": '1,2,3a,1p,2s,3b',
"print_warning": True, # 开启警告以便调试
"zs_algo": "normal", # 使用标准中枢算法提高稳定性 over_seg, normal, auto
"bi_algo": "normal",
"seg_algo": "chan",
})
# 将数据设置到WebDataAPI
from DataAPI.WebDataAPI import WebDataAPI
WebDataAPI.set_data("WEB_DATA", df)
# 创建CChan实例,使用自定义数据源
chan = CChan(
code="WEB_DATA",
begin_time=None,
end_time=None,
data_src="custom:WebDataAPI.WebDataAPI",
lv_list=[kl_type],
config=config,
autype=AUTYPE.QFQ,
)
# 获取分析结果
kline_list = chan[kl_type]
# 提取笔列表
bi_list = []
if hasattr(kline_list, 'bi_list') and kline_list.bi_list:
bi_list = kline_list.bi_list
# 提取线段列表
seg_list = []
if hasattr(kline_list, 'seg_list') and kline_list.seg_list:
seg_list = kline_list.seg_list
# 提取中枢列表
zs_list = []
if hasattr(kline_list, 'zs_list') and kline_list.zs_list:
zs_list = kline_list.zs_list
# 直接从KLine_List获取买卖点
buy_sell_points = []
try:
if hasattr(kline_list, 'bs_point_lst') and kline_list.bs_point_lst:
bsp_list = sorted(kline_list.bs_point_lst.lst, key=lambda x: x.klu.time)
for i, bsp in enumerate(bsp_list):
# 根据买卖点类型选择正确的价格
if bsp.is_buy:
# 买点使用低点价格
price = bsp.klu.low
else:
# 卖点使用高点价格
price = bsp.klu.high
if bsp.type2str().__contains__("1"):
buy_sell_points.append({
'type': 1 if bsp.is_buy else -1, # 简化买卖点类型
'time': bsp.klu.time, # 保持CTime对象,后续统一格式化
'price': price,
'desc': f"{bsp.type2str()}"
})
else:
buy_sell_points = []
except Exception as e:
print(f"[{timeframe}] 获取买卖点失败: {e}")
buy_sell_points = []
# 提取分型信息
klc_fx_info = []
klu_fx_info = []
# 从K线列表中提取分型信息
if hasattr(kline_list, 'lst'):
for klc in kline_list.lst:
if hasattr(klc, 'fx') and klc.fx != FX_TYPE.UNKNOWN:
klc_fx_info.append({
'time': klc.time_end,
'price': klc.low if klc.fx == FX_TYPE.BOTTOM else klc.high,
'fx_type': str(klc.fx).replace("FX_TYPE.", ""),
'is_bottom': klc.fx == FX_TYPE.BOTTOM,
'fx_strength': 1, # 基础分型强度
'fx_strength_level': "中",
'is_strong_fx': False
})
# 提取KLU分型信息
if hasattr(klc, 'lst'):
for klu in klc.lst:
if hasattr(klu, 'fx') and klu.fx != FX_TYPE.UNKNOWN:
klu_fx_info.append({
'time': klu.time,
'price': klu.low if klu.fx == FX_TYPE.BOTTOM else klu.high,
'fx_type': str(klu.fx).replace("FX_TYPE.", ""),
'is_bottom': klu.fx == FX_TYPE.BOTTOM,
'fx_strength': 1,
'fx_strength_level': "中",
'is_strong_fx': False,
'fx_confirmed': True
})
# 只在数据量较少时打印分析结果摘要,避免在回放时产生过多日志
if len(df) <= 50:
print(f"chan.py分析完成: 笔{len(bi_list)}个, 线段{len(seg_list)}个, 中枢{len(zs_list)}个, 买卖点{len(buy_sell_points)}个")
#last_seg = seg_list[-2]
#last_bi = last_seg.end_bi
#last_klu = last_bi.get_end_klu()
#print(last_klu.time, last_seg.is_sure, last_seg.dir, "--------------------------------")
return {
'klc_list': kline_list.lst if hasattr(kline_list, 'lst') else [],
'klu_list': [], # KLU数据在klc中
'bi_list': bi_list,
'seg_list': seg_list,
'zs_list': zs_list,
'trade_points': buy_sell_points,
'klc_fx_info': klc_fx_info,
'klu_fx_info': klu_fx_info
}
except Exception as e:
print(f"chan.py分析出错: {e}")
traceback.print_exc()
# 返回空结果
return {
'klc_list': [],
'klu_list': [],
'bi_list': [],
'seg_list': [],
'zs_list': [],
'trade_points': [],
'klc_fx_info': [],
'klu_fx_info': []
}
# 辅助函数,转换缠论方向枚举为整数
def convert_direction(direction):
"""转换方向枚举为数字"""
if direction == BI_DIR.UP or direction == SEG_DIR.UP:
return 1
elif direction == BI_DIR.DOWN or direction == SEG_DIR.DOWN:
return -1
else:
return 0
def format_time_safely(time_obj, client_tz):
"""安全地格式化时间对象,处理字符串、datetime和CTime对象"""
if time_obj is None:
return None
# 如果是CTime对象,转换为字符串
if hasattr(time_obj, 'ts'): # CTime对象有ts属性
from datetime import datetime
dt = datetime.fromtimestamp(time_obj.ts)
return dt.astimezone(client_tz).isoformat()
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对象
try:
return time_obj.astimezone(client_tz).isoformat()
except:
return str(time_obj)
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() == '':
print(f"错误: 空交易对")
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'
print(f"API请求参数: symbol={symbol}, timeframe={timeframe}, element_timeframe={element_timeframe}")
print(f"时间范围: start_time={start_time}, end_time={end_time}")
print(f"elements_only参数: 原始值={elements_only_param}, 处理后={elements_only}")
# 验证小周期是否小于主周期
if element_timeframe and not is_smaller_or_equal_timeframe(element_timeframe, timeframe):
print(f"错误: 元素周期 {element_timeframe} 大于主周期 {timeframe}")
return jsonify({'error': '分形元素时间周期必须小于或等于主图表时间周期'})
# 获取数据
df = get_kl_data(symbol, timeframe, start_time=start_time, end_time=end_time)
if df is None:
print(f"错误: 获取数据失败 - symbol={symbol}, timeframe={timeframe}")
return jsonify({'error': '获取数据失败'})
if len(df) == 0:
print(f"错误: 所选时间范围内没有数据 - symbol={symbol}, timeframe={timeframe}")
return jsonify({'error': '所选时间范围内没有数据'})
# 使用客户端指定的时区
client_tz = timezone(client_timezone)
# 如果只需要分形元素数据而不需要主周期数据,则初始化一个空结果
result = {
'timezone': client_timezone
}
# 如果不是只需要分形元素数据,则添加主周期数据
if not elements_only:
print(f"处理主周期数据 (elements_only={elements_only})")
# 添加技术指标(包括布林带)
df = add_indicators(df)
# 进行缠论分析
analysis_result = analyze_chan(df, timeframe)
# 计算MACD
macd_data = calculate_macd(df)
# 计算笔的MACD背离值(用于显示)
bi_macd_divs = calculate_bi_macd_divergence(analysis_result['bi_list'])
# 直接使用小周期相同的方式获取买卖点
all_trade_points = analysis_result['trade_points']
print(f"主周期买卖点:共{len(all_trade_points)}个")
# 添加主周期分析结果到返回数据
result.update({
'kline_data': clean_dataframe_for_json(df).to_dict('records'),
'bi_list': [{
'start_time': format_time_safely(bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(bi.end_klc.time_end, client_tz) if bi.end_klc else None,
'start_price': bi.begin_klc.low if convert_direction(bi.dir) == 1 else bi.begin_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_divs.get(i, 0))
} for i, bi in enumerate(analysis_result['bi_list']) if bi.end_klc],
'seg_list': [{
'start_time': format_time_safely(seg.start_bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(seg.end_bi.end_klc.time_end, client_tz) if seg.end_bi else None,
'start_price': seg.start_bi.begin_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.begin_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': format_time_safely(zs.begin.time, client_tz),
'end_time': format_time_safely(zs.end.time, client_tz) if zs.end else None,
'zg': zs.high,
'zd': zs.low,
'is_sure': zs.is_sure # 添加中枢是否完成的标志
} for zs in analysis_result['zs_list'] if zs.end],
# 添加未完成中枢列表
'uncompleted_zs_list': [{
'start_time': format_time_safely(zs.begin.time, client_tz),
'end_time': None, # 未完成中枢没有结束时间
'zg': zs.high,
'zd': zs.low,
'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 all_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()
},
# 添加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']]
})
else:
print(f"只请求元素数据,跳过主周期数据处理 (elements_only={elements_only})")
# 如果有指定分形元素时间周期,获取小周期数据
if element_timeframe:
print(f"处理元素周期数据: {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, element_timeframe)
# 计算小周期MACD数据
element_macd_data = calculate_macd(element_df)
# 计算小周期笔的MACD背离值(用于显示)
element_bi_macd_divs = calculate_bi_macd_divergence(element_analysis['bi_list'])
# 直接使用小周期Chan.py的买卖点,不再添加自定义买卖点
element_all_trade_points = element_analysis['trade_points']
print(f"使用小周期Chan.py内置买卖点:共{len(element_all_trade_points)}个")
# 添加小周期分析结果到返回数据
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()
}
result['element_bi_list'] = [{
'start_time': format_time_safely(bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(bi.end_klc.time_end, client_tz) if bi.end_klc else None,
'start_price': bi.begin_klc.low if convert_direction(bi.dir) == 1 else bi.begin_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(element_bi_macd_divs.get(i, 0))
} for i, bi in enumerate(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': format_time_safely(seg.start_bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(seg.end_bi.end_klc.time_end, client_tz) if seg.end_bi else None,
'start_price': seg.start_bi.begin_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.begin_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': format_time_safely(zs.begin.time, client_tz),
'end_time': format_time_safely(zs.end.time, client_tz) if zs.end else None,
'zg': zs.high,
'zd': zs.low,
'is_sure': zs.is_sure # 添加中枢是否完成的标志
} for zs in element_analysis['zs_list'] if zs.end]
result['element_uncompleted_zs_list'] = [{
'start_time': format_time_safely(zs.begin.time, client_tz),
'end_time': None, # 未完成中枢没有结束时间
'zg': zs.high,
'zd': zs.low,
'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_all_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']]
print(f"小周期分析完成: {element_timeframe}, 笔数量: {len(result['element_bi_list'])}, {'仅元素数据' if elements_only else '包含主周期数据'}")
else:
print(f"无法获取小周期数据: {element_timeframe}")
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/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:
print("正在获取完整股票列表...")
stock_list = china_stock.get_stock_list()
if stock_list and len(stock_list) > 0:
print(f"成功获取完整股票列表: {len(stock_list)} 只股票")
data_source = "完整股票列表"
else:
raise Exception("获取到的股票列表为空")
except Exception as e:
print(f"获取完整股票列表失败: {e}")
print("使用热门股票列表作为备用...")
try:
popular_stocks = china_stock.get_popular_stocks()
stock_list = [{'symbol': stock['symbol'], 'name': stock['name']} for stock in popular_stocks]
print(f"使用热门股票列表: {len(stock_list)} 只股票")
data_source = "热门股票列表"
except Exception as e2:
print(f"获取热门股票列表也失败: {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
print(f"开始筛选股票,总数: {total_count}, 时间范围: {start_time}{end_time}, 周期: {timeframe}")
for stock in stock_list:
try:
symbol = stock['symbol']
name = stock['name']
processed_count += 1
# 每处理20只股票打印一次进度
if processed_count % 20 == 0:
print(f"已处理 {processed_count}/{total_count} 只股票,成功: {len(results)}, 失败: {failed_count}")
# 获取股票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:
print(f"连续失败 {failed_count} 次,可能是网络问题")
return jsonify({
'error': '网络连接不稳定,无法获取股票数据。请检查网络连接后重试。',
'error_type': 'network_error',
'processed_count': processed_count,
'failed_count': failed_count
})
continue
# 进行缠论分析
analysis_result = analyze_chan(df, timeframe)
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:
print(f"处理股票 {symbol} 时出错: {str(e)}")
failed_count += 1
continue
print(f"筛选完成,共找到 {len(results)} 只满足条件的股票")
# 按分型强度降序排列
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:
print(f"筛选股票时发生错误: {str(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)
def calculate_bi_macd_divergence(bi_list):
"""计算每个笔的MACD背离值 - 使用CBi内置的Cal_MACD_area方法"""
bi_macd_divs = {}
if len(bi_list) < 4:
return bi_macd_divs
for i in range(3, len(bi_list)):
current_bi = bi_list[i]
prev_same_dir_bi = None
# 找到同方向的前一个笔
for j in range(i-2, -1, -1):
if convert_direction(bi_list[j].dir) == convert_direction(current_bi.dir):
prev_same_dir_bi = bi_list[j]
break
if not prev_same_dir_bi or not current_bi.end_klc or not prev_same_dir_bi.end_klc:
continue
try:
# 使用CBi内置的Cal_MACD_area方法计算MACD面积
current_macd_area = current_bi.Cal_MACD_area()
prev_macd_area = prev_same_dir_bi.Cal_MACD_area()
if current_macd_area <= 0 or prev_macd_area <= 0:
continue
# 计算价格变化
current_price = current_bi.get_end_val()
prev_price = prev_same_dir_bi.get_end_val()
# 计算背驰度:价格创新高/低但MACD面积没有创新高/低
divergence_value = 0
if convert_direction(current_bi.dir) == -1: # 向下笔
# 底背驰:价格创新低但MACD面积没有创新高(向下笔MACD面积越大表示力度越大)
if current_price < prev_price and current_macd_area < prev_macd_area:
divergence_value = (prev_macd_area - current_macd_area) / prev_macd_area
elif convert_direction(current_bi.dir) == 1: # 向上笔
# 顶背驰:价格创新高但MACD面积没有创新高
if current_price > prev_price and current_macd_area < prev_macd_area:
divergence_value = (prev_macd_area - current_macd_area) / prev_macd_area
# 存储背离值(使用笔的索引作为key)
bi_macd_divs[i] = divergence_value
except Exception as e:
# 如果计算出错,跳过这个笔
print(f"计算笔{i}的MACD背离时出错: {e}")
continue
return bi_macd_divs
@app.route('/api/replay_data')
def get_replay_data():
"""生成回放数据接口 - 为每个时间点计算当时可用的数据"""
symbol = request.args.get('symbol', 'SOL/USDT:USDT')
timeframe = request.args.get('timeframe', '5m')
# 验证交易对不为空
if not symbol or symbol.strip() == '':
print(f"错误: 空交易对")
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')
print(f"回放数据API请求参数: symbol={symbol}, timeframe={timeframe}, element_timeframe={element_timeframe}")
print(f"时间范围: start_time={start_time}, end_time={end_time}")
# 验证小周期是否小于主周期
if element_timeframe and not is_smaller_or_equal_timeframe(element_timeframe, timeframe):
print(f"错误: 元素周期 {element_timeframe} 大于主周期 {timeframe}")
return jsonify({'error': '分形元素时间周期必须小于或等于主图表时间周期'})
# 获取完整数据
df = get_kl_data(symbol, timeframe, start_time=start_time, end_time=end_time)
if df is None:
print(f"错误: 获取数据失败 - symbol={symbol}, timeframe={timeframe}")
return jsonify({'error': '获取数据失败'})
if len(df) == 0:
print(f"错误: 所选时间范围内没有数据 - symbol={symbol}, timeframe={timeframe}")
return jsonify({'error': '所选时间范围内没有数据'})
# 使用客户端指定的时区
client_tz = timezone(client_timezone)
# 预先获取小周期数据(如果需要的话),避免在循环中重复拉取
element_df = None
if element_timeframe:
element_df = get_kl_data(symbol, element_timeframe, start_time=start_time, end_time=end_time)
# 生成回放数据
replay_data = {}
print(f"开始生成回放数据,主周期: {len(df)} 条记录")
# 为每个时间点计算当时可用的数据
for i in range(len(df)):
# 获取到当前时间点的数据子集
current_df = df.iloc[:i+1].copy()
# 添加技术指标(包括布林带)
current_df = add_indicators(current_df)
# 进行缠论分析
analysis_result = analyze_chan(current_df, timeframe)
# 计算MACD
macd_data = calculate_macd(current_df)
# 计算笔的MACD背离值
bi_macd_divs = calculate_bi_macd_divergence(analysis_result['bi_list'])
# 获取买卖点
all_trade_points = analysis_result['trade_points']
# 构建当前时间点的数据
current_data = {
'kline_data': clean_dataframe_for_json(current_df).to_dict('records'),
'bi_list': [{
'start_time': format_time_safely(bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(bi.end_klc.time_end, client_tz) if bi.end_klc else None,
'start_price': bi.begin_klc.low if convert_direction(bi.dir) == 1 else bi.begin_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_divs.get(idx, 0))
} for idx, bi in enumerate(analysis_result['bi_list']) if bi.end_klc],
'seg_list': [{
'start_time': format_time_safely(seg.start_bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(seg.end_bi.end_klc.time_end, client_tz) if seg.end_bi else None,
'start_price': seg.start_bi.begin_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.begin_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': format_time_safely(zs.begin.time, client_tz),
'end_time': format_time_safely(zs.end.time, client_tz) if zs.end else None,
'zg': zs.high,
'zd': zs.low,
'is_sure': zs.is_sure
} for zs in analysis_result['zs_list'] if zs.end],
'uncompleted_zs_list': [{
'start_time': format_time_safely(zs.begin.time, client_tz),
'end_time': None,
'zg': zs.high,
'zd': zs.low,
'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 all_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()
},
'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 element_timeframe and element_df is not None and len(element_df) > 0:
# 找到对应当前主周期时间点的小周期数据截止位置
current_time = current_df['date'].iloc[-1]
element_subset = element_df[element_df['date'] <= current_time]
if len(element_subset) > 0:
# 添加小周期技术指标
element_subset = add_indicators(element_subset)
# 对小周期数据进行缠论分析
element_analysis = analyze_chan(element_subset, element_timeframe)
# 计算小周期MACD数据
element_macd_data = calculate_macd(element_subset)
# 计算小周期笔的MACD背离值
element_bi_macd_divs = calculate_bi_macd_divergence(element_analysis['bi_list'])
# 获取小周期买卖点
element_all_trade_points = element_analysis['trade_points']
# 添加小周期数据到当前时间点
current_data.update({
'element_timeframe': element_timeframe,
'element_macd': element_macd_data,
'element_bollinger': {
'upper': element_subset['bb_upper'].tolist(),
'middle': element_subset['bb_middle'].tolist(),
'lower': element_subset['bb_lower'].tolist()
},
'element_element_bollinger': {
'upper': element_subset['element_bb_upper'].tolist(),
'middle': element_subset['element_bb_middle'].tolist(),
'lower': element_subset['element_bb_lower'].tolist()
},
'element_bi_list': [{
'start_time': format_time_safely(bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(bi.end_klc.time_end, client_tz) if bi.end_klc else None,
'start_price': bi.begin_klc.low if convert_direction(bi.dir) == 1 else bi.begin_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(element_bi_macd_divs.get(idx, 0))
} for idx, bi in enumerate(element_analysis['bi_list']) if bi.end_klc],
'element_kline_data': clean_dataframe_for_json(element_subset).to_dict('records'),
'element_seg_list': [{
'start_time': format_time_safely(seg.start_bi.begin_klc.time_end, client_tz),
'end_time': format_time_safely(seg.end_bi.end_klc.time_end, client_tz) if seg.end_bi else None,
'start_price': seg.start_bi.begin_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.begin_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],
'element_zs_list': [{
'start_time': format_time_safely(zs.begin.time, client_tz),
'end_time': format_time_safely(zs.end.time, client_tz) if zs.end else None,
'zg': zs.high,
'zd': zs.low,
'is_sure': zs.is_sure
} for zs in element_analysis['zs_list'] if zs.end],
'element_uncompleted_zs_list': [{
'start_time': format_time_safely(zs.begin.time, client_tz),
'end_time': None,
'zg': zs.high,
'zd': zs.low,
'is_sure': zs.is_sure
} for zs in element_analysis['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 element_all_trade_points],
'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']],
'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']]
})
replay_data[i] = current_data
# 打印进度(减少频率)
if (i + 1) % 100 == 0 or i == len(df) - 1:
percent = int((i + 1) / len(df) * 100)
print(f"回放数据生成进度: {percent}% ({i + 1}/{len(df)})")
result = {
'replay_data': replay_data,
'total_length': len(df),
'timezone': client_timezone
}
print(f"回放数据生成完成: {len(df)} 条记录")
return jsonify(result)
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=8120)