jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replay is ok now

Browse Source
This commit is contained in:
jackyu66git
2025-06-10 18:46:42 +08:00
parent 2295a0b1f8
commit 747f74c1d1
3 changed files with 316 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLC.py
+3 -3
View File
@@ -160,7 +160,7 @@ class ChanKLC():
klu_list.append(klc3.klus)
gap = klc3.end_klu.index - klc1.start_klu.index + 1
if gap < 4:
print(klc1.start_time, klc1.start_klu.index, klc3.end_time, klc3.end_klu.index, gap, self.cal_fx_strength(), self.klc_fx_type)
pass
return gap
def get_feature_data(self):
features = dict()
@@ -1214,7 +1214,7 @@ class ChanKLC():
# 分型质量调整
base_score += fx_quality
print(self.start_time, base_score, is_bi_end, post_fx_confirmation, fx_quality)
#print(self.start_time, base_score, is_bi_end, post_fx_confirmation, fx_quality)
# 2025-06-07 08:15:00 1.5 0 0.8 0.7
# 限制在-3到3范围内
return max(-3, min(3, base_score))
@@ -1245,7 +1245,7 @@ class ChanKLC():
if len(subsequent_klcs) < 2:
return 0
print(subsequent_klcs[len(subsequent_klcs)-1].start_time)
pass
if self.fx == Chan_FX_TYPE.TOP:
return self._check_top_bi_ending(subsequent_klcs)
else: # BOTTOM
web/app.py
+306 -210
View File
@@ -80,7 +80,6 @@ def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
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):
@@ -92,7 +91,7 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
try:
since = int(start_time)
except ValueError:
print(f"无效的起始时间: {start_time}")
pass
# 结束时间处理
until = None
@@ -100,7 +99,7 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
try:
until = int(end_time)
except ValueError:
print(f"无效的结束时间: {end_time}")
pass
# 根据时间周期调整每次请求的数据量
batch_size = 1000 # 默认批次大小
@@ -121,45 +120,36 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
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:
# 尝试增加时间跳过可能的问题时间点
@@ -172,7 +162,6 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
# 数据为空的情况
if not all_ohlcv or len(all_ohlcv) == 0:
print(f"未获取到数据: {symbol}, {timeframe}")
return None
# 转换为DataFrame
@@ -192,26 +181,18 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
# 限制数据条数的逻辑 - 优先考虑时间范围
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):
@@ -256,22 +237,16 @@ def get_a_stock_kl_data(symbol, timeframe, limit=1000, start_time=None, 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):
@@ -353,13 +328,152 @@ def analyze_chan(df):
for bi in bi_list:
bi.cal_macd_div()
#print(bi.start_time, bi.macd_hist, bi.macd_div)
def generate_replay_data(df, client_tz):
"""生成逐步计算的回放数据"""
print(f"开始生成回放数据，K线总数: {len(df)}")
# 获取原始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线数据
pass
except Exception as 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
# 统一使用cal_fx_strength函数
if hasattr(klc, 'cal_fx_strength'):
fx_strength = klc.cal_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:
# 如果出错，仍然添加基本信息，但分型强度为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:
# 如果出错，仍然添加基本信息，但分型强度为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
})
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 generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, start_time=None, end_time=None):
"""生成逐步计算的回放数据"""
replay_data = {}
# 预先获取和分析完整的次周期数据（避免重复计算）
element_full_data = None
element_analysis_full = None
if element_timeframe and symbol:
# 一次性获取完整的次周期数据
element_full_data = get_kl_data(symbol, element_timeframe, start_time=start_time, end_time=end_time)
if element_full_data is not None and len(element_full_data) > 0:
# 一次性添加技术指标和进行缠论分析
element_full_data = add_indicators(element_full_data)
element_analysis_full = analyze_chan(element_full_data)
# 为每个K线索引计算分析结果
for i in range(1, len(df) + 1): # 从1开始，至少需要1根K线
try:
@@ -375,6 +489,161 @@ def generate_replay_data(df, client_tz):
# 计算MACD
macd_data = calculate_macd(current_df)
# 如果有次周期数据，筛选对应时间范围的数据
element_step_data = {}
if element_full_data is not None and element_analysis_full is not None:
# 获取当前主周期时间范围
current_end_time = current_df['timestamp'].iloc[-1] if len(current_df) > 0 else None
if current_end_time:
# 筛选次周期数据：只取时间戳小于等于当前主周期结束时间的数据
element_current_df = element_full_data[element_full_data['timestamp'] <= current_end_time].copy()
if len(element_current_df) > 0:
# 筛选对应的分析结果
def filter_by_time(items, time_attr='end_time'):
"""根据时间筛选分析结果"""
filtered = []
for item in items:
try:
if hasattr(item, time_attr):
item_time = getattr(item, time_attr)
if item_time:
if isinstance(item_time, str):
item_timestamp = pd.to_datetime(item_time).timestamp() * 1000
else:
item_timestamp = item_time.timestamp() * 1000
if item_timestamp <= current_end_time:
filtered.append(item)
elif hasattr(item, 'end_klc') and item.end_klc:
item_time = item.end_klc.end_time
if item_time:
if isinstance(item_time, str):
item_timestamp = pd.to_datetime(item_time).timestamp() * 1000
else:
item_timestamp = item_time.timestamp() * 1000
if item_timestamp <= current_end_time:
filtered.append(item)
except:
continue
return filtered
# 筛选笔、线段、中枢数据
filtered_bi_list = filter_by_time(element_analysis_full['bi_list'])
filtered_seg_list = filter_by_time(element_analysis_full['seg_list'])
filtered_zs_list = filter_by_time(element_analysis_full['zs_list'])
# 筛选买卖点（基于字典格式）
filtered_trade_points = []
for point in element_analysis_full['trade_points']:
try:
point_time = point['time']
if isinstance(point_time, str):
point_timestamp = pd.to_datetime(point_time).timestamp() * 1000
else:
point_timestamp = point_time.timestamp() * 1000
if point_timestamp <= current_end_time:
filtered_trade_points.append(point)
except:
continue
# 筛选分型信息
def filter_fx_info(fx_list):
filtered = []
for fx in fx_list:
try:
fx_time = fx['time']
if isinstance(fx_time, str):
fx_timestamp = pd.to_datetime(fx_time).timestamp() * 1000
else:
fx_timestamp = fx_time.timestamp() * 1000
if fx_timestamp <= current_end_time:
filtered.append(fx)
except:
continue
return filtered
filtered_klc_fx = filter_fx_info(element_analysis_full['klc_fx_info'])
filtered_klu_fx = filter_fx_info(element_analysis_full['klu_fx_info'])
# 计算当前时间范围的MACD
element_macd_data = calculate_macd(element_current_df)
element_step_data = {
'element_kline_data': clean_dataframe_for_json(element_current_df).to_dict('records'),
'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 filtered_bi_list if bi.end_klc],
'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 filtered_seg_list if seg.end_bi],
'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 filtered_zs_list if zs.end_klc],
'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
} for zs in filtered_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 filtered_trade_points],
'element_macd': element_macd_data,
'element_bollinger': {
'upper': element_current_df['bb_upper'].tolist(),
'middle': element_current_df['bb_middle'].tolist(),
'lower': element_current_df['bb_lower'].tolist()
},
'element_element_bollinger': {
'upper': element_current_df['element_bb_upper'].tolist(),
'middle': element_current_df['element_bb_middle'].tolist(),
'lower': element_current_df['element_bb_lower'].tolist()
},
'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 filtered_klc_fx],
'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 filtered_klu_fx]
}
# 构建该索引对应的分析结果
step_data = {
'kline_data': clean_dataframe_for_json(current_df).to_dict('records'),
@@ -447,158 +716,21 @@ def generate_replay_data(df, client_tz):
} for point in analysis_result['klu_fx_info']]
}
# 合并次周期数据到step_data中
step_data.update(element_step_data)
replay_data[i-1] = step_data # 使用0-based索引
# 每处理100个点输出一次进度
if i % 100 == 0 or i == len(df):
print(f"生成回放数据进度: {i}/{len(df)}")
except Exception as e:
print(f"生成第{i}步回放数据时出错: {e}")
continue
print(f"回放数据生成完成，总步数: {len(replay_data)}")
return replay_data
# 获取原始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
# 统一使用cal_fx_strength函数
if hasattr(klc, 'cal_fx_strength'):
fx_strength = klc.cal_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)} 个中枢")
# 1. 基于笔的二三类买卖点识别（更及时）
trade_points.extend(identify_bi_trade_points(bi_list, zs_list))
@@ -615,7 +747,6 @@ def identify_trade_points(bi_list, seg_list, zs_list):
# 按时间排序
trade_points.sort(key=lambda x: x['time'])
print(f"总共识别出 {len(trade_points)} 个买卖点")
return trade_points
def identify_bi_trade_points(bi_list, zs_list):
@@ -911,7 +1042,6 @@ def analyze():
# 验证交易对不为空
if not symbol or symbol.strip() == '':
print(f"错误: 空交易对")
return jsonify({'error': '交易对不能为空'})
# 获取时间范围参数
@@ -931,24 +1061,16 @@ def analyze():
# 获取是否需要回放数据的参数
need_replay_data = request.args.get('need_replay_data', 'false').lower() == '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}")
print(f"need_replay_data参数: {need_replay_data}")
# 验证小周期是否小于主周期
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': '所选时间范围内没有数据'})
# 使用客户端指定的时区
@@ -961,8 +1083,6 @@ def analyze():
# 如果不是只需要分形元素数据，则添加主周期数据
if not elements_only:
print(f"处理主周期数据 (elements_only={elements_only})")
# 添加技术指标（包括布林带）
df = add_indicators(df)
@@ -974,9 +1094,7 @@ def analyze():
# 如果需要回放数据，生成逐步计算的回放数据
if need_replay_data:
print("开始生成回放数据...")
replay_data = generate_replay_data(df, client_tz)
print(f"回放数据生成完成，包含 {len(replay_data)} 个步骤")
replay_data = generate_replay_data(df, client_tz, symbol, element_timeframe, start_time, end_time)
else:
replay_data = None
@@ -1058,12 +1176,9 @@ def analyze():
# 如果生成了回放数据，添加到返回结果中
if replay_data is not None:
result['replay_data'] = replay_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)
@@ -1160,9 +1275,7 @@ def analyze():
'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}")
pass
return jsonify(result)
@@ -1248,23 +1361,17 @@ def filter_stocks():
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({
@@ -1287,18 +1394,12 @@ def filter_stocks():
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)
@@ -1306,7 +1407,6 @@ def filter_stocks():
failed_count += 1
# 如果连续失败太多，可能是网络问题
if failed_count > 10 and len(results) == 0:
print(f"连续失败 {failed_count} 次，可能是网络问题")
return jsonify({
'error': '网络连接不稳定，无法获取股票数据。请检查网络连接后重试。',
'error_type': 'network_error',
@@ -1357,12 +1457,9 @@ def filter_stocks():
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)
@@ -1376,7 +1473,6 @@ def filter_stocks():
})
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({
web/cn_stock_data.py
+7 -34
View File
@@ -25,7 +25,7 @@ class ChinaStockData:
# 设置较短的超时时间，避免长时间等待
import akshare as ak
print("正在获取A股股票列表...")
pass
# 尝试获取沪深A股实时行情，设置超时时间
try:
@@ -42,12 +42,11 @@ class ChinaStockData:
delattr(requests, 'timeout')
except Exception as network_error:
print(f"网络请求失败: {network_error}")
pass
# 网络失败时返回空列表，让调用方使用备用方案
return []
if stock_info is None or len(stock_info) == 0:
print("获取到的股票数据为空")
return []
# 增加到前2000只股票，提供更多选择
@@ -66,16 +65,13 @@ class ChinaStockData:
'amount': float(row['成交额']) if pd.notna(row['成交额']) else 0.0
})
except Exception as row_error:
print(f"处理股票数据行时出错: {row_error}")
continue
# 按成交金额排序，优先显示活跃股票
stock_list.sort(key=lambda x: x['amount'], reverse=True)
print(f"成功获取 {len(stock_list)} 只股票")
return stock_list
except Exception as e:
print(f"获取股票列表失败: {e}")
return []
def get_popular_stocks(self):
@@ -224,7 +220,7 @@ class ChinaStockData:
if '-' in end_date:
end_date = end_date.replace('-', '')
print(f"获取A股数据: {symbol}, 周期: {timeframe}, 开始: {start_date}, 结束: {end_date}")
pass
# 分批次获取数据以突破单次限制
all_data = []
@@ -252,7 +248,7 @@ class ChinaStockData:
current_end_dt = current_start_dt + timedelta(days=batch_days)
current_end = min(current_end_dt.strftime('%Y%m%d'), end_date)
print(f"批次 {iteration_count}: 获取 {current_start} 到 {current_end} 的数据")
pass
try:
# 根据时间周期选择不同的API
@@ -294,13 +290,11 @@ class ChinaStockData:
df_batch = self.adjust_timestamp_for_trading_hours(df_batch, timeframe)
all_data.append(df_batch)
print(f"批次 {iteration_count}: 获取到 {len(df_batch)} 条记录")
else:
print(f"批次 {iteration_count}: 未获取到数据")
pass
except Exception as e:
print(f"批次 {iteration_count} 获取失败: {e}")
# 继续下一个批次
pass
# 更新下一批次的开始时间
current_start = (current_end_dt + timedelta(days=1)).strftime('%Y%m%d')
@@ -310,7 +304,6 @@ class ChinaStockData:
# 合并所有批次的数据
if not all_data:
print(f"未获取到任何数据: {symbol}")
return None
# 合并DataFrame
@@ -330,17 +323,13 @@ class ChinaStockData:
# 检查是否指定了明确的时间范围
if start_date and end_date:
# 如果指定了时间范围，优先返回完整的时间范围数据
print(f"用户指定了时间范围 {start_date} 到 {end_date}，返回完整数据 {len(df)} 条记录")
if len(df) > 10000: # 防止数据量过大，设置一个合理的上限
print(f"警告：数据量过大({len(df)}条)，为保证性能将限制为最新的10000条记录")
df = df.tail(10000).reset_index(drop=True)
else:
# 如果没有指定时间范围，使用默认的limit限制
print(f"未指定明确时间范围，应用默认限制，返回最新的 {limit} 条记录")
df = df.tail(limit).reset_index(drop=True)
elif limit is None and len(df) > 10000:
# 即使没有limit限制，也要防止数据量过大影响性能
print(f"无limit限制但数据量过大({len(df)}条)，为保证性能将限制为最新的10000条记录")
df = df.tail(10000).reset_index(drop=True)
# 添加技术指标
@@ -351,7 +340,6 @@ class ChinaStockData:
# 检查并处理任何剩余的NaN值
if df.isnull().any().any():
print("警告：发现NaN值，正在清理...")
# 对于数值列，用0填充NaN
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
@@ -367,12 +355,9 @@ class ChinaStockData:
for col in df.select_dtypes(include=[np.number]).columns:
df[col] = df[col].replace([np.inf, -np.inf], 0 if col != 'volume_ratio' else 1.0)
print(f"成功获取A股数据: {len(df)} 条记录 (共 {len(all_data)} 个批次)")
return df
except Exception as e:
print(f"获取A股数据失败: {e}")
traceback.print_exc()
return None
def add_indicators(self, df):
@@ -417,7 +402,6 @@ class ChinaStockData:
return df
except Exception as e:
print(f"添加指标失败: {e}")
return df
def search_stock(self, keyword):
@@ -478,7 +462,7 @@ class ChinaStockData:
break
except Exception as e:
print(f"全市场搜索失败: {e}")
pass
# 排序：优先显示代码匹配的结果
def sort_key(item):
@@ -495,7 +479,6 @@ class ChinaStockData:
return results[:20]
except Exception as e:
print(f"搜索股票失败: {e}")
return []
def get_stock_by_sector(self, sector=None):
@@ -514,7 +497,6 @@ class ChinaStockData:
sectors[sector_name].append(stock)
return sectors
except Exception as e:
print(f"获取行业股票失败: {e}")
return {} if sector is None else []
def get_all_sectors(self):
@@ -527,7 +509,6 @@ class ChinaStockData:
sectors.add(sector)
return sorted(list(sectors))
except Exception as e:
print(f"获取行业分类失败: {e}")
return []
def is_trading_day(self, date):
@@ -547,7 +528,6 @@ class ChinaStockData:
# 目前暂时只过滤周末
return True
except Exception as e:
print(f"判断交易日失败: {e}")
return True # 默认返回True，避免过度过滤
def is_trading_time(self, dt):
@@ -569,7 +549,6 @@ class ChinaStockData:
return ((morning_start <= time_str <= morning_end) or
(afternoon_start <= time_str <= afternoon_end))
except Exception as e:
print(f"判断交易时间失败: {e}")
return True # 默认返回True，避免过度过滤
def adjust_timestamp_for_trading_hours(self, df, timeframe):
@@ -605,8 +584,6 @@ class ChinaStockData:
return df.reset_index(drop=True)
except Exception as e:
print(f"调整A股时间戳失败: {e}")
traceback.print_exc()
return df
def get_trading_calendar(self, start_date, end_date):
@@ -627,7 +604,6 @@ class ChinaStockData:
return trading_days
except Exception as e:
print(f"获取交易日历失败: {e}")
# 如果获取失败，生成简单的工作日列表（排除周末）
trading_days = []
current = pd.to_datetime(start_date)
@@ -721,8 +697,6 @@ class ChinaStockData:
return df
except Exception as e:
print(f"填补交易时间间隙失败: {e}")
traceback.print_exc()
return df
def clean_a_stock_data(self, df, timeframe):
@@ -778,5 +752,4 @@ class ChinaStockData:
return df.reset_index(drop=True)
except Exception as e:
print(f"清理A股数据失败: {e}")
return df