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

refactor: 缠论引擎包化与 Web 分层（ECR-001）

Browse Source 
将根目录引擎迁入 chanlun/ 并保留兼容 shim；拆分 TF_DF 与 web 服务；
前端模块化；strategies 改用 chanlun 导入；补充 ESS 文档与 golden 回归。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
2026-08-05 18:48:20 +08:00
co-authored by Cursor
parent e2e45bc1bc
commit 74dec4e50b
160 changed files with 25576 additions and 23104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/realtime_fx_example.py
+222
View File
@@ -0,0 +1,222 @@
import sys, os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
实时K线分型强弱判断示例
解决KLC滞后问题，提供即时的分型信号
"""
from chanlun.core.ChanKLU import ChanKLU
from chanlun.core.ChanEnum import Chan_FX_TYPE
import pandas as pd
from datetime import datetime, timedelta
class RealtimeFxAnalyzer:
"""实时分型分析器"""
def __init__(self):
self.klu_list = []
self.latest_signals = []
def add_kline(self, time, open_price, high, low, close, volume, indicators=None):
"""
添加新的K线数据并进行实时分析
Args:
time: 时间
open_price, high, low, close, volume: K线数据
indicators: 技术指标字典 {'macd': xx, 'rsi': xx, 'ma5': xx, ...}
"""
# 创建新的KLU对象
new_klu = ChanKLU(time, open_price, high, low, close, volume)
# 设置技术指标
if indicators:
new_klu.set_indicators(indicators)
# 设置索引
new_klu.set_idx(len(self.klu_list))
# 建立前后关系链
if len(self.klu_list) >= 1:
prev_klu = self.klu_list[-1]
new_klu.set_pre(prev_klu)
prev_klu.set_next(new_klu)
# 如果有足够的数据，设置前一根K线的next关系
if len(self.klu_list) >= 2:
prev_prev_klu = self.klu_list[-2]
prev_prev_klu.set_next(self.klu_list[-1])
self.klu_list.append(new_klu)
# 实时分析最近的K线分型
self._analyze_recent_fractals()
return new_klu
def _analyze_recent_fractals(self):
"""分析最近的分型情况"""
if len(self.klu_list) < 3:
return
# 检查倒数第二根K线的分型（因为需要左右两根K线确认）
target_idx = len(self.klu_list) - 2
if target_idx >= 1:
target_klu = self.klu_list[target_idx]
# 进行实时分型分析
target_klu.update_realtime_analysis()
# 如果发现分型，记录信号
if target_klu.fx_confirmed:
signal = target_klu.get_fx_signal()
signal_info = {
'time': target_klu.time,
'price': target_klu.close,
'signal_type': signal[0],
'strength': signal[1],
'suggestion': signal[2],
'fx_type': target_klu.fx_type
}
self.latest_signals.append(signal_info)
# 保持最近20个信号
if len(self.latest_signals) > 20:
self.latest_signals.pop(0)
print(f"🔔 分型信号: {signal_info['time']} - {signal_info['signal_type']} "
f"(强度: {signal_info['strength']}) - {signal_info['suggestion']}")
def get_latest_signal(self):
"""获取最新的分型信号"""
return self.latest_signals[-1] if self.latest_signals else None
def get_current_fx_status(self):
"""获取当前分型状态统计"""
if len(self.klu_list) < 10:
return {"status": "数据不足"}
recent_10 = self.klu_list[-10:]
top_fx_count = sum(1 for klu in recent_10 if klu.fx_type == Chan_FX_TYPE.TOP)
bottom_fx_count = sum(1 for klu in recent_10 if klu.fx_type == Chan_FX_TYPE.BOTTOM)
strong_fx_count = sum(1 for klu in recent_10 if klu.fx_strength >= 65)
return {
"最近10根K线": len(recent_10),
"顶分型数量": top_fx_count,
"底分型数量": bottom_fx_count,
"强分型数量": strong_fx_count,
"最新K线时间": recent_10[-1].time,
"最新信号": self.get_latest_signal()
}
def simulate_realtime_trading():
"""模拟实时交易场景"""
print("=== 实时K线分型分析示例 ===\n")
# 创建分析器
analyzer = RealtimeFxAnalyzer()
# 模拟实时K线数据流
base_time = datetime.now()
base_price = 100.0
print("开始接收K线数据...\n")
for i in range(20):
# 模拟价格波动
if i < 5: # 上涨阶段
price_change = 0.5
elif i < 10: # 下跌阶段
price_change = -0.8
elif i < 15: # 震荡阶段
price_change = 0.3 * ((-1) ** i)
else: # 再次上涨
price_change = 0.6
current_price = base_price + price_change
# 构造K线数据
open_price = base_price
high = max(open_price, current_price) + abs(price_change) * 0.2
low = min(open_price, current_price) - abs(price_change) * 0.2
close = current_price
volume = 1000 + i * 50
# 模拟技术指标
indicators = {
'ma5': base_price + (i - 10) * 0.1,
'ma10': base_price + (i - 10) * 0.05,
'rsi': 50 + (i % 7 - 3) * 10,
'macd': (i % 6 - 3) * 0.01,
'macdhist': (i % 4 - 2) * 0.005,
'volume_ratio': 1.0 + (i % 3 - 1) * 0.2
}
# 添加K线数据
kline_time = base_time + timedelta(minutes=i)
analyzer.add_kline(
time=kline_time.strftime("%Y-%m-%d %H:%M:%S"),
open_price=open_price,
high=high,
low=low,
close=close,
volume=volume,
indicators=indicators
)
base_price = current_price
# 每5根K线显示一次状态
if (i + 1) % 5 == 0:
status = analyzer.get_current_fx_status()
print(f"\n--- 第{i+1}根K线后的状态 ---")
for key, value in status.items():
if key != "最新信号":
print(f"{key}: {value}")
if "最新信号" in status and status["最新信号"]:
signal = status["最新信号"]
print(f"最新信号: {signal['signal_type']} (强度: {signal['strength']})")
print()
print("\n=== 所有分型信号汇总 ===")
for signal in analyzer.latest_signals:
print(f"{signal['time']} | {signal['signal_type']} | 强度: {signal['strength']} | {signal['suggestion']}")
def compare_latency():
"""对比KLC和KLU方法的延迟差异"""
print("\n=== 延迟对比分析 ===")
print("假设场景：连续包含关系的K线序列")
print("原始K线: K1, K2(包含K1), K3(包含K2), K4(突破), K5, K6")
print()
print("KLC方法:")
print("- 需要等待K4确认包含关系结束")
print("- KLC1 = [K1+K2+K3], 在K4完成时才确定")
print("- 分型检测: 需要等待KLC1, KLC2, KLC3")
print("- 实际延迟: 可能6-8根原始K线")
print()
print("KLU实时方法:")
print("- 每根K线完成时立即检测")
print("- K3完成时就能检测K2的分型状态")
print("- 实际延迟: 最多1根K线")
print()
print("延迟改善: 从6-8根K线缩短到1根K线")
print("时间价值: 在5分钟K线下，可节省25-40分钟的反应时间")
if __name__ == "__main__":
# 运行模拟
simulate_realtime_trading()
# 显示延迟对比
compare_latency()