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

Change to the new added

Browse Source
This commit is contained in:
jackyu66git
2025-06-12 17:02:15 +08:00
parent 747f74c1d1
commit b4ded5d7d2
3 changed files with 178 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLC.py
+3 -3
View File
@@ -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
# 可能笔终结：部分条件满足
ChanKLU.py
+2 -2
View File
@@ -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
web/app.py
+173 -75
View File
@@ -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股股票"""