diff --git a/ChanKLC.py b/ChanKLC.py index d7ab700..9d4b490 100644 --- a/ChanKLC.py +++ b/ChanKLC.py @@ -1217,7 +1217,7 @@ class ChanKLC(): #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)) + return base_score def _check_if_bi_ending_fx(self, klc_offset): """ @@ -1280,7 +1280,7 @@ class ChanKLC(): downward_trend += 1 # 强烈笔终结：跌破关键位且无新高 - if broken_key_levels >= 1 and new_highs == 0 and downward_trend >= 2: + if broken_key_levels >= 1 and new_highs == 0 and downward_trend >= 1: return 2 # 可能笔终结：部分条件满足 @@ -1326,7 +1326,7 @@ class ChanKLC(): upward_trend += 1 # 强烈笔终结：突破关键位且无新低 - if broken_key_levels >= 1 and new_lows == 0 and upward_trend >= 2: + if broken_key_levels >= 1 and new_lows == 0 and upward_trend >= 1: return 2 # 可能笔终结：部分条件满足 diff --git a/ChanKLU.py b/ChanKLU.py index 0a030a9..5531dee 100644 --- a/ChanKLU.py +++ b/ChanKLU.py @@ -269,8 +269,8 @@ class ChanKLU: self.fx_strength = final_score # 转换为0-100分制以保持接口一致性 #self.fx_strength = int((final_score + 3) * 100 / 6) # -3到3映射到0-100 - if final_score > 1.8: - print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality) + #if final_score > 1.8: + #print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality) #print(self.time, final_score, is_bi_end, post_fx_confirmation, fx_quality) #self.fx_strength = self.cal_fx() return self.fx_strength diff --git a/web/app.py b/web/app.py index 9691a71..1697060 100644 --- a/web/app.py +++ b/web/app.py @@ -463,16 +463,14 @@ def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, sta """生成逐步计算的回放数据""" 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线 @@ -491,7 +489,7 @@ def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, sta # 如果有次周期数据，筛选对应时间范围的数据 element_step_data = {} - if element_full_data is not None and element_analysis_full is not None: + if element_full_data is not None: # 获取当前主周期时间范围 current_end_time = current_df['timestamp'].iloc[-1] if len(current_df) > 0 else None @@ -500,75 +498,17 @@ def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, sta 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 + # 重新对当前时间范围的次周期数据进行缠论分析 + # 这样可以确保数据的准确性，避免时间筛选的复杂性 + element_current_analysis = analyze_chan(element_current_df) - # 筛选笔、线段、中枢数据 - 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']) + # 直接使用分析结果，无需复杂的时间筛选 + filtered_bi_list = element_current_analysis['bi_list'] + filtered_seg_list = element_current_analysis['seg_list'] + filtered_zs_list = element_current_analysis['zs_list'] + filtered_trade_points = element_current_analysis['trade_points'] + filtered_klc_fx = element_current_analysis['klc_fx_info'] + filtered_klu_fx = element_current_analysis['klu_fx_info'] # 计算当前时间范围的MACD element_macd_data = calculate_macd(element_current_df) @@ -646,6 +586,10 @@ def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, sta # 构建该索引对应的分析结果 step_data = { + 'step_index': i-1, # 当前步骤索引 + 'total_steps': len(df), # 总步骤数 + 'has_element_data': element_timeframe is not None and len(element_step_data) > 0, # 是否包含次周期数据 + 'element_timeframe': element_timeframe, # 次周期时间框架 'kline_data': clean_dataframe_for_json(current_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(), @@ -716,8 +660,24 @@ def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, sta } for point in analysis_result['klu_fx_info']] } - # 合并次周期数据到step_data中 - step_data.update(element_step_data) + # 合并次周期数据到step_data中，如果没有次周期数据则提供空的占位符 + if element_step_data: + step_data.update(element_step_data) + else: + # 提供空的次周期数据结构，确保前端可以统一处理 + step_data.update({ + 'element_kline_data': [], + 'element_bi_list': [], + 'element_seg_list': [], + 'element_zs_list': [], + 'element_uncompleted_zs_list': [], + 'element_trade_points': [], + 'element_macd': {'macd': [], 'signal': [], 'histogram': []}, + 'element_bollinger': {'upper': [], 'middle': [], 'lower': []}, + 'element_element_bollinger': {'upper': [], 'middle': [], 'lower': []}, + 'element_klc_fx_info': [], + 'element_klu_fx_info': [] + }) replay_data[i-1] = step_data # 使用0-based索引 @@ -1344,6 +1304,144 @@ def search_stock(): except Exception as e: return jsonify({'error': str(e)}) +@app.route('/api/test_element_data') +def test_element_data(): + """测试次周期数据是否正确生成""" + try: + symbol = request.args.get('symbol', 'SOL/USDT:USDT') + timeframe = request.args.get('timeframe', '1h') + element_timeframe = request.args.get('element_timeframe', '15m') + + # 获取主周期数据 + main_df = get_kl_data(symbol, timeframe, limit=3) + if main_df is None or len(main_df) == 0: + return jsonify({'error': '无法获取主周期数据'}) + + # 获取次周期数据 + element_df = get_kl_data(symbol, element_timeframe, + start_time=main_df['timestamp'].iloc[0], + end_time=main_df['timestamp'].iloc[-1]) + + if element_df is None or len(element_df) == 0: + return jsonify({'error': '无法获取次周期数据'}) + + # 分析次周期数据 + element_df = add_indicators(element_df) + element_analysis = analyze_chan(element_df) + + return jsonify({ + 'main_data_count': len(main_df), + 'element_data_count': len(element_df), + 'element_analysis': { + 'bi_count': len(element_analysis['bi_list']), + 'seg_count': len(element_analysis['seg_list']), + 'zs_count': len(element_analysis['zs_list']), + 'klc_fx_count': len(element_analysis['klc_fx_info']), + 'klu_fx_count': len(element_analysis['klu_fx_info']), + 'trade_points_count': len(element_analysis['trade_points']) + }, + 'sample_bi': [{'has_end_klc': bi.end_klc is not None, + 'direction': convert_direction(bi.dir)} + for bi in element_analysis['bi_list'][:2]] if len(element_analysis['bi_list']) > 0 else [], + 'sample_klc_fx': element_analysis['klc_fx_info'][:3] if len(element_analysis['klc_fx_info']) > 0 else [], + 'sample_klu_fx': element_analysis['klu_fx_info'][:3] if len(element_analysis['klu_fx_info']) > 0 else [] + }) + + except Exception as e: + import traceback + return jsonify({'error': str(e), 'traceback': traceback.format_exc()}) + +@app.route('/api/debug_replay_sample') +def debug_replay_sample(): + """调试接口：返回回放数据样本，方便前端调试""" + try: + symbol = request.args.get('symbol', 'SOL/USDT:USDT') + timeframe = request.args.get('timeframe', '1h') + element_timeframe = request.args.get('element_timeframe', '15m') + step = int(request.args.get('step', 2)) # 返回第几步的数据 + + # 获取少量数据进行测试 + df = get_kl_data(symbol, timeframe, limit=5) + if df is None or len(df) == 0: + return jsonify({'error': '无法获取测试数据'}) + + # 生成回放数据 + client_tz = timezone('Asia/Shanghai') + replay_data = generate_replay_data( + df, client_tz, symbol, element_timeframe, + start_time=None, end_time=None + ) + + if step not in replay_data: + return jsonify({'error': f'步骤 {step} 不存在，可用步骤：{list(replay_data.keys())}'}) + + # 返回指定步骤的完整数据 + step_data = replay_data[step] + + return jsonify({ + 'step': step, + 'data': step_data, + 'summary': { + 'has_element_data': step_data.get('has_element_data', False), + 'element_timeframe': step_data.get('element_timeframe'), + 'main_bi_count': len(step_data.get('bi_list', [])), + 'main_klc_fx_count': len(step_data.get('klc_fx_info', [])), + 'main_klu_fx_count': len(step_data.get('klu_fx_info', [])), + 'element_bi_count': len(step_data.get('element_bi_list', [])), + 'element_klc_fx_count': len(step_data.get('element_klc_fx_info', [])), + 'element_klu_fx_count': len(step_data.get('element_klu_fx_info', [])), + 'element_kline_count': len(step_data.get('element_kline_data', [])) + } + }) + + except Exception as e: + return jsonify({'error': str(e)}) + +@app.route('/api/debug_replay_structure') +def debug_replay_structure(): + """调试接口：检查回放数据结构""" + try: + # 获取一个简单的测试案例 + symbol = request.args.get('symbol', 'SOL/USDT:USDT') + timeframe = request.args.get('timeframe', '1h') + element_timeframe = request.args.get('element_timeframe', '15m') + + # 获取少量数据进行测试 + df = get_kl_data(symbol, timeframe, limit=5) # 只取5根K线 + if df is None or len(df) == 0: + return jsonify({'error': '无法获取测试数据'}) + + # 生成回放数据 + client_tz = timezone('Asia/Shanghai') + replay_data = generate_replay_data( + df, client_tz, symbol, element_timeframe, + start_time=None, end_time=None + ) + + # 返回结构信息 + result = { + 'total_steps': len(replay_data), + 'sample_step_keys': list(replay_data[0].keys()) if len(replay_data) > 0 else [], + 'has_element_data_in_steps': [], + 'element_data_counts': {} + } + + # 检查每个步骤的次周期数据 + for step_idx, step_data in replay_data.items(): + has_element = step_data.get('has_element_data', False) + result['has_element_data_in_steps'].append({ + 'step': step_idx, + 'has_element_data': has_element, + 'element_bi_count': len(step_data.get('element_bi_list', [])), + 'element_klc_fx_count': len(step_data.get('element_klc_fx_info', [])), + 'element_klu_fx_count': len(step_data.get('element_klu_fx_info', [])) + }) + + return jsonify(result) + + except Exception as e: + return jsonify({'error': str(e)}) + @app.route('/api/filter_stocks', methods=['POST']) def filter_stocks(): """筛选满足条件的A股股票"""