220 lines
7.4 KiB
Python
220 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
实时K线分型强弱判断示例
|
|
解决KLC滞后问题，提供即时的分型信号
|
|
"""
|
|
|
|
from ChanKLU import ChanKLU
|
|
from 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() |