Files
Chan/web/app.py
T
2025-06-02 14:34:28 +08:00

1123 lines
48 KiB
Python

from flask import Flask, render_template, jsonify, request
import ccxt
import pandas as pd
from datetime import datetime, timedelta
import sys
import os
import matplotlib
matplotlib.use('Agg') # 设置使用非GUI后端,必须在导入pyplot之前设置
import matplotlib.pyplot as plt
import io
import base64
import time
import traceback
from pytz import timezone
import talib.abstract as ta
import numpy as np
# 添加父目录到系统路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ChanLun import ChanLun
from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE
from cn_stock_data import ChinaStockData
# 添加买卖点枚举类型
class TRADE_POINT_TYPE:
BUY1 = 1 # 一类买点
BUY2 = 2 # 二类买点
BUY3 = 3 # 三类买点
SELL1 = -1 # 一类卖点
SELL2 = -2 # 二类卖点
SELL3 = -3 # 三类卖点
app = Flask(__name__)
# 初始化交易所
exchange = ccxt.binance({
'enableRateLimit': True,
})
# 初始化A股数据获取器
china_stock = ChinaStockData()
# 时间周期映射
TIMEFRAMES = {
'1m': '1分钟',
'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'
]
# A股热门股票
A_STOCK_SYMBOLS = china_stock.get_popular_stocks()
def detect_symbol_type(symbol):
"""检测交易对类型:crypto 或 a_stock"""
if '/' in symbol and 'USDT' in symbol:
return 'crypto'
elif len(symbol) == 6 and symbol.isdigit():
return 'a_stock'
else:
return 'unknown'
def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
"""获取K线数据,支持加密货币和A股"""
symbol_type = detect_symbol_type(symbol)
if symbol_type == 'crypto':
return get_crypto_kl_data(symbol, timeframe, limit, start_time, end_time)
elif symbol_type == 'a_stock':
return get_a_stock_kl_data(symbol, timeframe, limit, start_time, end_time)
else:
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']:
batch_size = 1000
else:
batch_size = 1500 # 日线及以上可以获取更多
# 初始化存储所有K线数据的列表
all_ohlcv = []
# 初始化当前查询的开始时间
current_since = since
# 添加请求计数和最大限制
request_count = 0
max_requests = 50 # 最大请求次数,防止无限循环
print(f"开始分批获取加密货币数据: {symbol}, {timeframe}")
# 分页加载数据
while request_count < max_requests:
request_count += 1
print(f"批次 {request_count}: 获取数据 since={current_since}, limit={batch_size}")
try:
# 获取当前页的数据
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=current_since, limit=batch_size)
# 如果没有获取到数据,结束循环
if not ohlcv or len(ohlcv) == 0:
print(f"批次 {request_count}: 未获取到数据,结束")
break
# 将获取到的数据添加到总列表中
all_ohlcv.extend(ohlcv)
print(f"批次 {request_count}: 获取到 {len(ohlcv)} 条记录")
# 获取最后一条数据的时间戳
last_timestamp = ohlcv[-1][0]
# 如果已达到结束时间,结束循环
if until and last_timestamp >= until:
print(f"批次 {request_count}: 已达到结束时间,结束")
break
# 如果获取的数据条数小于限制数,说明已经获取完所有数据
if len(ohlcv) < batch_size:
print(f"批次 {request_count}: 数据不足批次大小,已获取完所有数据")
break
# 更新下一页的开始时间(加1毫秒避免重复)
current_since = last_timestamp + 1
except Exception as e:
print(f"批次 {request_count} 获取失败: {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:
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')
# 限制数据条数的逻辑 - 优先考虑时间范围
if start_time and end_time:
# 如果指定了明确的时间范围,返回该时间范围内的所有数据
print(f"用户指定了时间范围,返回完整数据 {len(df)} 条记录")
if len(df) > 10000: # 防止数据量过大,设置一个合理的上限
print(f"警告:数据量过大({len(df)}条),为保证性能将限制为最新的10000条记录")
df = df.tail(10000).reset_index(drop=True)
elif limit and len(df) > limit:
# 如果没有指定明确时间范围,使用默认的limit限制
print(f"未指定明确时间范围,应用默认限制,返回最新的 {limit} 条记录")
df = df.tail(limit).reset_index(drop=True)
# 如果过滤后没有数据,返回None
if len(df) == 0:
print("过滤后无数据")
return None
print(f"成功获取加密货币数据: {len(df)} 条记录 (共 {request_count} 个批次)")
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:
print(f"用户指定了时间范围 {start_date}{end_date},将获取该范围内的所有数据")
actual_limit = None # 不限制数据条数,获取完整时间范围数据
# 调用A股数据获取器
df = china_stock.get_kl_data(symbol, timeframe, start_date, end_date, actual_limit)
if df is None:
print(f"未获取到A股数据: {symbol}")
return None
print(f"获取到A股数据: {len(df)} 条记录")
return df
except Exception as e:
print(f"获取A股数据错误: {e}")
traceback.print_exc()
return None
def add_indicators(df):
fast = 8
slow = 16
period = 6
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
df['macd'] = macd['macd']
df['macdsignal'] = macd['macdsignal']
df['macdhist'] = macd['macdhist']
df['ma5'] = (ta.MA(df, timeperiod=5)).fillna(0)
df['ma10'] = (ta.MA(df, timeperiod=10)).fillna(0)
df['ma30'] = (ta.EMA(df, timeperiod=30)).fillna(0)
df['ma250'] = (ta.MA(df, timeperiod=250)).fillna(0)
df['rsi'] = ta.RSI(df, timeperiod=14)
# 计算布林带 (当前周期 - 20周期,2标准差)
bb = ta.BBANDS(df, timeperiod=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):
"""进行缠论分析"""
chan = ChanLun()
# 获取分析结果
klc_list = chan.get_klc_list(df)
bi_list = chan.cal_bi_list(klc_list)
#for index in range(0, 10):
#print(bi_list[index].start_time, bi_list[index].start_klc.end_time, bi_list[index].dir)
seg_list = chan.get_seg_list(bi_list)
zs_list = chan.calculate_zs(bi_list, seg_list)
# 添加买卖点识别
buy_sell_points = identify_trade_points(bi_list, seg_list, zs_list)
for bi in bi_list:
bi.cal_macdhist()
for bi in bi_list:
bi.cal_macd_div()
#print(bi.start_time, bi.macd_hist, bi.macd_div)
# 获取原始K线数据用于KLU分型分析
klu_list = []
try:
# 尝试获取KLU数据
if hasattr(chan, 'get_klu_list'):
klu_list = chan.get_klu_list(df)
elif hasattr(chan, 'klu_list'):
klu_list = chan.klu_list
else:
# 如果没有专门的KLU方法,尝试从KLC获取原始K线数据
print("未找到KLU数据获取方法,尝试其他方式")
except Exception as e:
print(f"获取KLU数据时出错: {e}")
klu_list = []
# 提取K线分型信息
klc_fx_info = []
for klc in klc_list:
if hasattr(klc, 'klc_fx_type') and klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
try:
# 计算分型强度
fx_strength = 0
fx_strength_level = ""
is_strong_fx = False
# 尝试调用分型强度计算方法
if hasattr(klc, 'cal_fx_strength'):
fx_strength = klc.cal_fx_strength()
elif hasattr(klc, 'calculate_fx_strength'):
fx_strength = klc.calculate_fx_strength()
# 尝试获取分型强度等级
if hasattr(klc, 'get_fx_strength_level'):
fx_strength_level = klc.get_fx_strength_level()
# 尝试判断是否为强分型
if hasattr(klc, 'is_strong_fx'):
is_strong_fx = klc.is_strong_fx()
# 如果分型强度小于1,设为0
if fx_strength < 1:
fx_strength = 0
klc_fx_info.append({
'time': klc.end_time,
'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high,
'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""),
'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM,
'fx_strength': fx_strength, # 分型强度分数 (0-100)
'fx_strength_level': fx_strength_level, # 分型强度等级 (极强/强/中等/弱/极弱)
'is_strong_fx': is_strong_fx # 是否为强分型
})
except Exception as e:
print(f"处理KLC分型信息时出错: {e}")
# 如果出错,仍然添加基本信息,但分型强度为0
klc_fx_info.append({
'time': klc.end_time,
'price': klc.low if klc.fx == Chan_FX_TYPE.BOTTOM else klc.high,
'fx_type': str(klc.klc_fx_type).replace("Chan_KLC_FX.", ""),
'is_bottom': klc.fx == Chan_FX_TYPE.BOTTOM,
'fx_strength': 0,
'fx_strength_level': "",
'is_strong_fx': False
})
# 提取KLU分型信息
klu_fx_info = []
for klu in klu_list:
if hasattr(klu, 'fx_type') and klu.fx_type != Chan_FX_TYPE.UNKNOWN:
try:
# 计算分型强度
fx_strength = 0
fx_strength_level = ""
is_strong_fx = False
# 尝试调用分型强度计算方法
if hasattr(klu, 'calculate_realtime_fx_strength'):
fx_strength = klu.calculate_realtime_fx_strength()
elif hasattr(klu, 'fx_strength'):
fx_strength = klu.fx_strength
# 尝试获取分型强度等级 - 基于强度值生成等级
if fx_strength >= 2:
fx_strength_level = "强"
is_strong_fx = True
elif fx_strength >= 1:
fx_strength_level = "中"
is_strong_fx = False
elif fx_strength >= 0:
fx_strength_level = "弱"
is_strong_fx = False
else:
fx_strength_level = "极弱"
is_strong_fx = False
# 确保分型确认状态
is_confirmed = getattr(klu, 'fx_confirmed', True)
klu_fx_info.append({
'time': klu.time,
'price': klu.low if klu.fx_type == Chan_FX_TYPE.BOTTOM else klu.high,
'fx_type': str(klu.fx_type).replace("Chan_FX_TYPE.", ""),
'is_bottom': klu.fx_type == Chan_FX_TYPE.BOTTOM,
'fx_strength': fx_strength, # 分型强度分数
'fx_strength_level': fx_strength_level, # 分型强度等级
'is_strong_fx': is_strong_fx, # 是否为强分型
'fx_confirmed': is_confirmed # 分型是否确认
})
except Exception as e:
print(f"处理KLU分型信息时出错: {e}")
# 如果出错,仍然添加基本信息,但分型强度为0
klu_fx_info.append({
'time': klu.time,
'price': klu.low if klu.fx_type == Chan_FX_TYPE.BOTTOM else klu.high,
'fx_type': str(klu.fx_type).replace("Chan_FX_TYPE.", ""),
'is_bottom': klu.fx_type == Chan_FX_TYPE.BOTTOM,
'fx_strength': 0,
'fx_strength_level': "",
'is_strong_fx': False,
'fx_confirmed': False
})
print(f"提取到 {len(klc_fx_info)} 个KLC分型和 {len(klu_fx_info)} 个KLU分型")
return {
'klc_list': klc_list,
'klu_list': klu_list, # 添加KLU列表
'bi_list': bi_list,
'seg_list': seg_list,
'zs_list': zs_list,
'trade_points': buy_sell_points,
'klc_fx_info': klc_fx_info, # KLC分型信息
'klu_fx_info': klu_fx_info # 添加KLU分型信息
}
def 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 or direction == Chan_SEG_DIR.UP:
return 1
elif direction == Chan_BI_DIR.DOWN or direction == Chan_SEG_DIR.DOWN:
return -1
else:
return 0
def format_time_safely(time_obj, client_tz):
"""安全地格式化时间对象,处理字符串和datetime两种情况"""
if time_obj is None:
return None
if isinstance(time_obj, str):
# 尝试将字符串解析为datetime
try:
from dateutil import parser
time_obj = parser.parse(time_obj)
return time_obj.astimezone(client_tz).isoformat()
except:
return time_obj
else:
# 已经是datetime对象
return time_obj.astimezone(client_tz).isoformat()
def is_smaller_timeframe(tf1, tf2):
"""判断时间周期tf1是否小于tf2"""
# 定义时间周期的分钟数映射
tf_values = {
'1m': 1,
'3m': 3,
'5m': 5,
'15m': 15,
'30m': 30,
'1h': 60,
'2h': 120,
'4h': 240,
'6h': 360,
'8h': 480,
'12h': 720,
'1d': 1440,
'3d': 4320,
'1w': 10080,
'1M': 43200
}
# 获取时间周期对应的分钟数
tf1_value = tf_values.get(tf1)
tf2_value = tf_values.get(tf2)
# 如果某个时间周期不在映射中,返回False
if tf1_value is None or tf2_value is None:
return False
# 返回tf1是否小于tf2
return tf1_value < tf2_value
def is_smaller_or_equal_timeframe(tf1, tf2):
"""判断时间周期tf1是否小于等于tf2"""
# 定义时间周期的分钟数映射
tf_values = {
'1m': 1,
'3m': 3,
'5m': 5,
'15m': 15,
'30m': 30,
'1h': 60,
'2h': 120,
'4h': 240,
'6h': 360,
'8h': 480,
'12h': 720,
'1d': 1440,
'3d': 4320,
'1w': 10080,
'1M': 43200
}
# 获取时间周期对应的分钟数
tf1_value = tf_values.get(tf1)
tf2_value = tf_values.get(tf2)
# 如果某个时间周期不在映射中,返回False
if tf1_value is None or tf2_value is None:
return False
# 返回tf1是否小于等于tf2
return tf1_value <= tf2_value
def clean_dataframe_for_json(df):
"""清理DataFrame数据用于JSON序列化"""
# 创建副本避免修改原始数据
clean_df = df.copy()
# 替换NaN值为None
clean_df = clean_df.where(pd.notnull(clean_df), None)
return clean_df
@app.route('/')
def index():
"""主页"""
return render_template('index.html',
timeframes=TIMEFRAMES,
symbols=SYMBOLS,
a_stock_symbols=A_STOCK_SYMBOLS)
@app.route('/api/analyze')
def analyze():
"""分析接口"""
symbol = request.args.get('symbol', 'SOL/USDT:USDT')
timeframe = request.args.get('timeframe', '5m')
# 验证交易对不为空
if not symbol or symbol.strip() == '':
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)
# 计算MACD
macd_data = calculate_macd(df)
# 添加主周期分析结果到返回数据
result.update({
'kline_data': clean_dataframe_for_json(df).to_dict('records'),
'bi_list': [{
'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None,
'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None,
'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high,
'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None,
'direction': convert_direction(bi.dir),
'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0
} for bi in analysis_result['bi_list'] if bi.end_klc],
'seg_list': [{
'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None,
'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None,
'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high,
'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None,
'direction': convert_direction(seg.dir)
} for seg in analysis_result['seg_list'] if seg.end_bi],
'zs_list': [{
'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None,
'zg': zs.zg,
'zd': zs.zd,
'is_sure': zs.is_sure # 添加中枢是否完成的标志
} for zs in analysis_result['zs_list'] if zs.end_klc],
# 添加未完成中枢列表
'uncompleted_zs_list': [{
'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': None, # 未完成中枢没有结束时间
'zg': zs.zg,
'zd': zs.zd,
'is_sure': zs.is_sure # 未完成中枢的is_sure为False
} for zs in analysis_result['zs_list'] if not zs.is_sure],
'trade_points': [{
'type': point['type'],
'time': format_time_safely(point['time'], client_tz),
'price': point['price'],
'desc': point['desc']
} for point in analysis_result['trade_points']],
'macd': macd_data,
# 添加布林带数据
'bollinger': {
'upper': df['bb_upper'].tolist(),
'middle': df['bb_middle'].tolist(),
'lower': df['bb_lower'].tolist()
},
'element_bollinger': {
'upper': df['element_bb_upper'].tolist(),
'middle': df['element_bb_middle'].tolist(),
'lower': df['element_bb_lower'].tolist()
},
# 添加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)
# 计算小周期MACD数据
element_macd_data = calculate_macd(element_df)
# 添加小周期分析结果到返回数据
result['element_timeframe'] = element_timeframe
result['element_macd'] = element_macd_data # 添加小周期MACD数据
# 添加小周期布林带数据
result['element_bollinger'] = {
'upper': element_df['bb_upper'].tolist(),
'middle': element_df['bb_middle'].tolist(),
'lower': element_df['bb_lower'].tolist()
}
result['element_element_bollinger'] = {
'upper': element_df['element_bb_upper'].tolist(),
'middle': element_df['element_bb_middle'].tolist(),
'lower': element_df['element_bb_lower'].tolist()
}
result['element_bi_list'] = [{
'start_time': bi.start_klc.end_time if isinstance(bi.start_klc.end_time, str) else bi.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (bi.end_klc.end_time if isinstance(bi.end_klc.end_time, str) else bi.end_klc.end_time.astimezone(client_tz).isoformat()) if bi.end_klc else None,
'sure_time': format_time_safely(bi.sure_time, client_tz) if bi.sure_time else None,
'start_price': bi.start_klc.low if convert_direction(bi.dir) == 1 else bi.start_klc.high,
'end_price': bi.end_klc.high if convert_direction(bi.dir) == 1 else bi.end_klc.low if bi.end_klc else None,
'direction': convert_direction(bi.dir),
'macd_div': float(bi.macd_div) if hasattr(bi, 'macd_div') else 0
} for bi in element_analysis['bi_list'] if bi.end_klc]
# 添加小周期K线数据
result['element_kline_data'] = clean_dataframe_for_json(element_df).to_dict('records')
result['element_seg_list'] = [{
'start_time': seg.start_bi.start_klc.end_time if isinstance(seg.start_bi.start_klc.end_time, str) else seg.start_bi.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (seg.end_bi.end_klc.end_time if isinstance(seg.end_bi.end_klc.end_time, str) else seg.end_bi.end_klc.end_time.astimezone(client_tz).isoformat()) if seg.end_bi else None,
'sure_time': format_time_safely(seg.sure_time, client_tz) if seg.sure_time else None,
'start_price': seg.start_bi.start_klc.low if convert_direction(seg.dir) == 1 else seg.start_bi.start_klc.high,
'end_price': seg.end_bi.end_klc.high if convert_direction(seg.dir) == 1 else seg.end_bi.end_klc.low if seg.end_bi else None,
'direction': convert_direction(seg.dir)
} for seg in element_analysis['seg_list'] if seg.end_bi]
result['element_zs_list'] = [{
'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': (zs.end_klc.end_time if isinstance(zs.end_klc.end_time, str) else zs.end_klc.end_time.astimezone(client_tz).isoformat()) if zs.end_klc else None,
'zg': zs.zg,
'zd': zs.zd,
'is_sure': zs.is_sure # 添加中枢是否完成的标志
} for zs in element_analysis['zs_list'] if zs.end_klc]
result['element_uncompleted_zs_list'] = [{
'start_time': zs.start_klc.end_time if isinstance(zs.start_klc.end_time, str) else zs.start_klc.end_time.astimezone(client_tz).isoformat(),
'end_time': None, # 未完成中枢没有结束时间
'zg': zs.zg,
'zd': zs.zd,
'is_sure': zs.is_sure # 未完成中枢的is_sure为False
} for zs in element_analysis['zs_list'] if not zs.is_sure]
result['element_trade_points'] = [{
'type': point['type'],
'time': format_time_safely(point['time'], client_tz),
'price': point['price'],
'desc': point['desc']
} for point in element_analysis['trade_points']]
# 添加小周期分型信息
result['element_klc_fx_info'] = [{
'time': format_time_safely(point['time'], client_tz),
'price': float(point['price']),
'fx_type': point['fx_type'],
'is_bottom': bool(point['is_bottom']),
'fx_strength': float(point['fx_strength']), # 分型强度分数
'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级
'is_strong_fx': bool(point['is_strong_fx']) # 是否为强分型
} for point in element_analysis['klc_fx_info']]
result['element_klu_fx_info'] = [{
'time': format_time_safely(point['time'], client_tz),
'price': float(point['price']),
'fx_type': point['fx_type'],
'is_bottom': bool(point['is_bottom']),
'fx_strength': float(point['fx_strength']), # 分型强度分数
'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级
'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型
'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认
} for point in element_analysis['klu_fx_info']]
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)
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)
if __name__ == '__main__':
app.run(debug=True, host='0.0.0.0', port=8120)