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 # 添加父目录到系统路径 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 # 添加买卖点枚举类型 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分钟', '5m': '5分钟', '15m': '15分钟', '30m': '30分钟', '1h': '1小时', '4h': '4小时', '1d': '日线', '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' ] def get_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}") # 初始化存储所有K线数据的列表 all_ohlcv = [] # 初始化当前查询的开始时间 current_since = since # 分页加载数据 while True: print(f"获取数据: {symbol}, {timeframe}, limit={limit}, since={current_since}") # 获取当前页的数据 ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=current_since, limit=limit) # 如果没有获取到数据，结束循环 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) < limit: break # 更新下一页的开始时间（加1毫秒避免重复） current_since = last_timestamp + 1 # 防止API请求过于频繁 time.sleep(0.5) # 等待0.5秒 # 数据为空的情况 if not all_ohlcv or len(all_ohlcv) == 0: print(f"未获取到数据: {symbol}, {timeframe}") 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') chan = ChanLun() klc_list = chan.get_klc_list(df) klc_index = 0 df_copy = df.copy() ret_df = pd.DataFrame(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume', 'date']) for index in range(0, len(df)): if klc_index >= len(klc_list): klc_index = len(klc_list) - 1 klc = klc_list[klc_index] if index == klc.start_klu.index: ret_df.loc[klc_index] = df_copy.loc[index] klc_index += 1 # 如果过滤后没有数据，返回None if len(ret_df) == 0: print("过滤后无数据") return None print(f"获取到总共 {len(ret_df)} 条数据") return ret_df except Exception as e: print(f"获取数据错误: {e}") traceback.print_exc() return None 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): """进行缠论分析""" chan = ChanLun() # 获取分析结果 klc_list = chan.get_klc_list(df) bi_list = chan.cal_bi_list(klc_list) seg_list = chan.get_seg_list(bi_list) zs_list = chan.calculate_zs(bi_list, seg_list) # 添加买卖点识别 buy_sell_points = identify_trade_points(bi_list, seg_list, zs_list) return { 'klc_list': klc_list, 'bi_list': bi_list, 'seg_list': seg_list, 'zs_list': zs_list, 'trade_points': buy_sell_points } def identify_trade_points(bi_list, seg_list, zs_list): """识别缠论买卖点 - 只保留最重要的一类买卖点，减少标记干扰""" trade_points = [] # 输出调试信息 print(f"识别买卖点：总共 {len(bi_list)} 个笔, {len(seg_list)} 个线段, {len(zs_list)} 个中枢") # 只识别一类买卖点：线段向上或向下突破 if len(seg_list) >= 3: for i in range(2, len(seg_list)): # 确保线段已完成 if seg_list[i].end_bi and seg_list[i-1].end_bi and seg_list[i-2].end_bi: # 一类买点：向下-向上-向下的底分型，第三段结束点为买点 if (convert_direction(seg_list[i-2].dir) == -1 and convert_direction(seg_list[i-1].dir) == 1 and convert_direction(seg_list[i].dir) == -1): print(f"发现一类买点：线段方向 {convert_direction(seg_list[i-2].dir)}-{convert_direction(seg_list[i-1].dir)}-{convert_direction(seg_list[i].dir)}") trade_points.append({ 'type': TRADE_POINT_TYPE.BUY1, 'time': seg_list[i].end_bi.end_klc.end_time, 'price': seg_list[i].end_bi.end_klc.low, 'desc': '一类买点' }) # 一类卖点：向上-向下-向上的顶分型，第三段结束点为卖点 if (convert_direction(seg_list[i-2].dir) == 1 and convert_direction(seg_list[i-1].dir) == -1 and convert_direction(seg_list[i].dir) == 1): print(f"发现一类卖点：线段方向 {convert_direction(seg_list[i-2].dir)}-{convert_direction(seg_list[i-1].dir)}-{convert_direction(seg_list[i].dir)}") trade_points.append({ 'type': TRADE_POINT_TYPE.SELL1, 'time': seg_list[i].end_bi.end_klc.end_time, 'price': seg_list[i].end_bi.end_klc.high, 'desc': '一类卖点' }) print(f"总共识别出 {len(trade_points)} 个买卖点") return trade_points # 辅助函数，转换缠论方向枚举为整数 def convert_direction(direction): """将缠论方向枚举转换为整数""" if direction == Chan_BI_DIR.UP: return 1 elif direction == Chan_BI_DIR.DOWN: return -1 elif direction == Chan_SEG_DIR.UP: return 1 elif direction == Chan_SEG_DIR.DOWN: return -1 else: return 0 def format_time_safely(time_obj, client_tz): """安全地格式化时间对象，处理字符串和datetime两种情况""" if time_obj is None: return None if isinstance(time_obj, str): # 尝试将字符串解析为datetime try: from dateutil import parser time_obj = parser.parse(time_obj) return time_obj.astimezone(client_tz).isoformat() except: return time_obj else: # 已经是datetime对象 return time_obj.astimezone(client_tz).isoformat() def is_smaller_timeframe(tf1, tf2): """判断时间周期tf1是否小于tf2""" # 定义时间周期的分钟数映射 tf_values = { '1m': 1, '3m': 3, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '2h': 120, '4h': 240, '6h': 360, '8h': 480, '12h': 720, '1d': 1440, '3d': 4320, '1w': 10080, '1M': 43200 } # 获取时间周期对应的分钟数 tf1_value = tf_values.get(tf1) tf2_value = tf_values.get(tf2) # 如果某个时间周期不在映射中，返回False if tf1_value is None or tf2_value is None: return False # 返回tf1是否小于tf2 return tf1_value < tf2_value def is_smaller_or_equal_timeframe(tf1, tf2): """判断时间周期tf1是否小于等于tf2""" # 定义时间周期的分钟数映射 tf_values = { '1m': 1, '3m': 3, '5m': 5, '15m': 15, '30m': 30, '1h': 60, '2h': 120, '4h': 240, '6h': 360, '8h': 480, '12h': 720, '1d': 1440, '3d': 4320, '1w': 10080, '1M': 43200 } # 获取时间周期对应的分钟数 tf1_value = tf_values.get(tf1) tf2_value = tf_values.get(tf2) # 如果某个时间周期不在映射中，返回False if tf1_value is None or tf2_value is None: return False # 返回tf1是否小于等于tf2 return tf1_value <= tf2_value @app.route('/') def index(): """主页""" return render_template('index.html', timeframes=TIMEFRAMES, symbols=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})") # 进行缠论分析 analysis_result = analyze_chan(df) # 计算MACD macd_data = calculate_macd(df) # 添加主周期分析结果到返回数据 result.update({ 'kline_data': df.to_dict('records'), 'bi_list': [{ 'start_time': bi.start_klc.start_time if isinstance(bi.start_klc.start_time, str) else bi.start_klc.start_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, '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) } for bi in analysis_result['bi_list'] if bi.end_klc], 'seg_list': [{ 'start_time': seg.start_bi.start_klc.start_time if isinstance(seg.start_bi.start_klc.start_time, str) else seg.start_bi.start_klc.start_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, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in analysis_result['seg_list'] if seg.end_bi], 'zs_list': [{ 'start_time': zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 添加中枢是否完成的标志 } for zs in analysis_result['zs_list'] if zs.end_klc], # 添加未完成中枢列表 'uncompleted_zs_list': [{ 'start_time': zs.start_klc.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, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in analysis_result['zs_list'] if not zs.is_sure], 'trade_points': [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in analysis_result['trade_points']], 'macd': macd_data }) 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_analysis = analyze_chan(element_df) # 添加小周期分析结果到返回数据 result['element_timeframe'] = element_timeframe result['element_bi_list'] = [{ 'start_time': bi.start_klc.start_time if isinstance(bi.start_klc.start_time, str) else bi.start_klc.start_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, '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) } for bi in element_analysis['bi_list'] if bi.end_klc] result['element_seg_list'] = [{ 'start_time': seg.start_bi.start_klc.start_time if isinstance(seg.start_bi.start_klc.start_time, str) else seg.start_bi.start_klc.start_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, 'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high, 'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None, 'direction': convert_direction(seg.dir) } for seg in element_analysis['seg_list'] if seg.end_bi] result['element_zs_list'] = [{ 'start_time': zs.start_klc.start_time if isinstance(zs.start_klc.start_time, str) else zs.start_klc.start_time.astimezone(client_tz).isoformat(), 'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None, 'zg': zs.zg, 'zd': zs.zd, 'is_sure': zs.is_sure # 添加中枢是否完成的标志 } for zs in element_analysis['zs_list'] if zs.end_klc] # 添加小周期未完成中枢列表 result['element_uncompleted_zs_list'] = [{ 'start_time': zs.start_klc.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, 'is_sure': zs.is_sure # 未完成中枢的is_sure为False } for zs in element_analysis['zs_list'] if not zs.is_sure] result['element_trade_points'] = [{ 'type': point['type'], 'time': format_time_safely(point['time'], client_tz), 'price': point['price'], 'desc': point['desc'] } for point in element_analysis['trade_points']] 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)}) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=8124)