440 lines
17 KiB
Python
440 lines
17 KiB
Python
import copy
|
|
from typing import Dict, Optional
|
|
|
|
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR
|
|
import ChanKLU
|
|
import ChanCTime
|
|
|
|
# 根据结合律合并K线后的K线
|
|
class ChanKLC():
|
|
def __init__(self, klu: ChanKLU, index, ddir=Chan_KLINE_DIR.UP):
|
|
self.start_time = klu.time
|
|
self.end_time = None
|
|
self.high = klu.high
|
|
self.low = klu.low
|
|
self.dir = ddir
|
|
self.index = index
|
|
self.klus = []
|
|
self.add_klu(klu)
|
|
self.fx = Chan_FX_TYPE.UNKNOWN
|
|
self.next = None
|
|
self.pre = None
|
|
self.start_klu = klu
|
|
self.end_klu = None
|
|
self.state = "00"
|
|
self.open = klu.open
|
|
self.close = klu.close
|
|
self.volume = klu.volume
|
|
def add_klu(self, klu):
|
|
self.klus.append(klu)
|
|
def set_end_klu(self, klu):
|
|
self.end_klu = klu
|
|
self.end_time = klu.time
|
|
self.close = klu.close
|
|
for index in range(1, len(self.klus)):
|
|
self.volume += self.klus[index].volume
|
|
def set_next(self, klc):
|
|
self.next = klc
|
|
def set_pre(self, klc):
|
|
self.pre = klc
|
|
def set_state(self, state):
|
|
self.state = state
|
|
def check_klu_included(self, klu):
|
|
if self.high >= klu.high:
|
|
# high大于，low小于，左包含
|
|
if self.low <= klu.low:
|
|
self.add_klu(klu=klu)
|
|
# gn>gn-1
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
# UP -> max(dn)
|
|
self.low = klu.low
|
|
else:
|
|
# DOWN -> min(gn)
|
|
self.high = klu.high
|
|
#self.print(klu, "Z")
|
|
return True
|
|
# high大于，low大于，不包含
|
|
else:
|
|
# if self.low > klu.low
|
|
# high相等，右包含
|
|
if self.high == klu.high:
|
|
self.add_klu(klu=klu)
|
|
# UP -> max(gn)
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
self.high = klu.high
|
|
else:
|
|
# DOWN -> min(dn)
|
|
self.low = klu.low
|
|
return True
|
|
else:
|
|
return False
|
|
else:
|
|
# high小于，low大于，右包含
|
|
if self.low >= klu.low:
|
|
self.add_klu(klu=klu)
|
|
# gn>gn-1
|
|
if self.dir == Chan_KLINE_DIR.UP:
|
|
# UP -> max(gn)
|
|
self.high = klu.high
|
|
else:
|
|
# DOWN -> min(dn)
|
|
self.low = klu.low
|
|
#self.print(klu, "Y")
|
|
return True
|
|
else:
|
|
# high小于，low小于，不包含
|
|
return False
|
|
def set_fx(self, fx: Chan_FX_TYPE):
|
|
self.fx = fx
|
|
def print(self):
|
|
print(self.time, self.high, self.low, self.start_time, self.end_time, self.fx, self.index)
|
|
def copy(self):
|
|
"""创建KLC对象的浅拷贝, 避免循环引用"""
|
|
new_klc = ChanKLC(self.start_klu, self.index, self.dir)
|
|
new_klc.high = self.high
|
|
new_klc.low = self.low
|
|
new_klc.state = self.state
|
|
new_klc.fx = self.fx
|
|
# 不复制 next 和 pre 引用，避免循环引用
|
|
return new_klc
|
|
def set_pre_fx(self):
|
|
if self.pre and self.pre.pre:
|
|
self.pre.fx = self.check_fx(self.pre.pre, self.pre)
|
|
def check_fx(self, k1, k2):
|
|
if k2.high > k1.high and k2.high > self.high:
|
|
return Chan_FX_TYPE.TOP
|
|
elif k2.low < k1.low and k2.low < self.low:
|
|
return Chan_FX_TYPE.BOTTOM
|
|
else:
|
|
return Chan_FX_TYPE.UNKNOWN
|
|
def set_bi_data(self, bi):
|
|
self.bi = bi
|
|
def cal_klu_features(self):
|
|
features = dict()
|
|
feature_sums = dict()
|
|
feature_counts = dict()
|
|
|
|
# 遍历所有klu，累计每个特征的总和和计数
|
|
for klu in self.klus:
|
|
for key, value in klu.get_feature_data().items():
|
|
if key not in feature_sums:
|
|
feature_sums[key] = 0
|
|
feature_counts[key] = 0
|
|
|
|
feature_sums[key] += value
|
|
feature_counts[key] += 1
|
|
|
|
# 计算每个特征的平均值
|
|
for key in feature_sums:
|
|
features[key] = feature_sums[key] / feature_counts[key]
|
|
|
|
return features
|
|
def get_feature_data(self):
|
|
features = dict()
|
|
# 原有基础特征
|
|
features['klc_close'] = self.close #0
|
|
features['klc_open'] = self.open #1
|
|
features['klc_high'] = self.high #2
|
|
features['klc_low'] = self.low #3
|
|
features['klc_index'] = self.index #4
|
|
features['klc_dir'] = 0 if self.dir == Chan_KLINE_DIR.UP else 1 #5
|
|
features['klc_state'] = self.state #6
|
|
features['klc_fx'] = 0 if self.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.fx == Chan_FX_TYPE.TOP else 2 #7
|
|
features['klc_klus'] = len(self.klus) #8
|
|
features['klc_volume'] = self.volume #9
|
|
features['klc_pre_fx'] = (0 if self.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre else 0
|
|
|
|
# ===== 2.1 K线形态因子 =====
|
|
|
|
# K线实体大小
|
|
if self.open != 0: # 避免除以零
|
|
features['klc_body_size_rel'] = abs(self.close - self.open) / self.open # 相对实体大小
|
|
else:
|
|
features['klc_body_size_rel'] = 0
|
|
features['klc_body_size_abs'] = abs(self.close - self.open) # 绝对实体大小
|
|
|
|
# 上下影线长度
|
|
max_oc = max(self.open, self.close)
|
|
min_oc = min(self.open, self.close)
|
|
high_low_range = self.high - self.low
|
|
|
|
if high_low_range != 0: # 避免除以零
|
|
features['klc_upper_shadow'] = (self.high - max_oc) / high_low_range # 上影线相对长度
|
|
features['klc_lower_shadow'] = (min_oc - self.low) / high_low_range # 下影线相对长度
|
|
else:
|
|
features['klc_upper_shadow'] = 0
|
|
features['klc_lower_shadow'] = 0
|
|
|
|
# K线波动范围
|
|
if self.close != 0: # 避免除以零
|
|
features['klc_range'] = (self.high - self.low) / self.close
|
|
else:
|
|
features['klc_range'] = 0
|
|
|
|
# 与前K线的价格关系
|
|
if self.pre:
|
|
# 当前K线最高价与前一根K线最高价的比较
|
|
if self.pre.high != 0: # 避免除以零
|
|
features['klc_high_ratio'] = self.high / self.pre.high
|
|
else:
|
|
features['klc_high_ratio'] = 1
|
|
|
|
# 当前K线最低价与前一根K线最低价的比较
|
|
if self.pre.low != 0: # 避免除以零
|
|
features['klc_low_ratio'] = self.low / self.pre.low
|
|
else:
|
|
features['klc_low_ratio'] = 1
|
|
|
|
# 当前K线收盘价与前一根K线收盘价的相对位置
|
|
if self.pre.close != 0: # 避免除以零
|
|
features['klc_close_change_1'] = (self.close - self.pre.close) / self.pre.close
|
|
else:
|
|
features['klc_close_change_1'] = 0
|
|
|
|
# 如果有前两根K线
|
|
if self.pre.pre:
|
|
if self.pre.pre.close != 0: # 避免除以零
|
|
features['klc_close_change_2'] = (self.close - self.pre.pre.close) / self.pre.pre.close
|
|
else:
|
|
features['klc_close_change_2'] = 0
|
|
else:
|
|
features['klc_close_change_2'] = 0
|
|
else:
|
|
# 如果没有前K线，设置默认值
|
|
features['klc_high_ratio'] = 1
|
|
features['klc_low_ratio'] = 1
|
|
features['klc_close_change_1'] = 0
|
|
features['klc_close_change_2'] = 0
|
|
|
|
# 分型特征编码
|
|
# 这里直接使用现有的fx字段，不重复计算
|
|
|
|
# ===== 2.2 价格关系因子 =====
|
|
|
|
# 价格与均线的关系 (从KLU中获取)
|
|
klu_features = self.cal_klu_features()
|
|
|
|
# MA5与收盘价的关系
|
|
if 'klu_ma5' in klu_features and klu_features['klu_ma5'] != 0:
|
|
features['klc_close_to_ma5'] = (self.close - klu_features['klu_ma5']) / klu_features['klu_ma5']
|
|
else:
|
|
features['klc_close_to_ma5'] = 0
|
|
|
|
# MA10与收盘价的关系
|
|
if 'klu_ma10' in klu_features and klu_features['klu_ma10'] != 0:
|
|
features['klc_close_to_ma10'] = (self.close - klu_features['klu_ma10']) / klu_features['klu_ma10']
|
|
else:
|
|
features['klc_close_to_ma10'] = 0
|
|
|
|
# MA30与收盘价的关系
|
|
if 'klu_ma30' in klu_features and klu_features['klu_ma30'] != 0:
|
|
features['klc_close_to_ma30'] = (self.close - klu_features['klu_ma30']) / klu_features['klu_ma30']
|
|
else:
|
|
features['klc_close_to_ma30'] = 0
|
|
|
|
# 短期均线与长期均线的差异
|
|
if 'klu_ma5' in klu_features and 'klu_ma30' in klu_features and klu_features['klu_ma30'] != 0:
|
|
features['klc_ma_diff'] = (klu_features['klu_ma5'] - klu_features['klu_ma30']) / klu_features['klu_ma30']
|
|
else:
|
|
features['klc_ma_diff'] = 0
|
|
|
|
# 价格突破特征
|
|
# 检查当前K线是否突破前3根K线的最高/最低价
|
|
if self.pre:
|
|
max_high = self.pre.high
|
|
min_low = self.pre.low
|
|
|
|
temp = self.pre
|
|
count = 1
|
|
while temp.pre and count < 3:
|
|
temp = temp.pre
|
|
max_high = max(max_high, temp.high)
|
|
min_low = min(min_low, temp.low)
|
|
count += 1
|
|
|
|
features['klc_break_high'] = 1 if self.high > max_high else 0
|
|
features['klc_break_low'] = 1 if self.low < min_low else 0
|
|
else:
|
|
features['klc_break_high'] = 0
|
|
features['klc_break_low'] = 0
|
|
|
|
# ===== 2.3 技术指标因子 =====
|
|
|
|
# 获取技术指标
|
|
# RSI (从KLU中获取)
|
|
if 'klu_rsi' in klu_features:
|
|
features['klc_rsi'] = klu_features['klu_rsi']
|
|
else:
|
|
features['klc_rsi'] = 50 # 默认中性值
|
|
|
|
# MACD (从KLU中获取)
|
|
if 'klu_macd' in klu_features:
|
|
features['klc_macd'] = klu_features['klu_macd']
|
|
else:
|
|
features['klc_macd'] = 0
|
|
|
|
if 'klu_signal' in klu_features:
|
|
features['klc_macd_signal'] = klu_features['klu_signal']
|
|
else:
|
|
features['klc_macd_signal'] = 0
|
|
|
|
if 'klu_macdhist' in klu_features:
|
|
features['klc_macd_hist'] = klu_features['klu_macdhist']
|
|
else:
|
|
features['klc_macd_hist'] = 0
|
|
|
|
# 成交量变化
|
|
if self.pre:
|
|
vol_sum = 0
|
|
count = 0
|
|
temp = self.pre
|
|
|
|
# 计算前5根K线的平均成交量
|
|
while temp and count < 5:
|
|
vol_sum += temp.volume
|
|
count += 1
|
|
temp = temp.pre
|
|
|
|
avg_vol = vol_sum / count if count > 0 else self.volume
|
|
|
|
if avg_vol != 0: # 避免除以零
|
|
features['klc_vol_ratio'] = self.volume / avg_vol
|
|
else:
|
|
features['klc_vol_ratio'] = 1
|
|
else:
|
|
features['klc_vol_ratio'] = 1
|
|
|
|
# ===== 2.4 市场环境因子 =====
|
|
|
|
# 价格波动率 (前5根K线收盘价的标准差)
|
|
if self.pre:
|
|
close_vals = [self.close]
|
|
temp = self.pre
|
|
count = 0
|
|
|
|
while temp and count < 5:
|
|
close_vals.append(temp.close)
|
|
count += 1
|
|
temp = temp.pre
|
|
|
|
if len(close_vals) > 1:
|
|
import numpy as np
|
|
std_dev = np.std(close_vals)
|
|
avg_close = np.mean(close_vals)
|
|
|
|
if avg_close != 0: # 避免除以零
|
|
features['klc_volatility'] = std_dev / avg_close
|
|
else:
|
|
features['klc_volatility'] = 0
|
|
else:
|
|
features['klc_volatility'] = 0
|
|
else:
|
|
features['klc_volatility'] = 0
|
|
|
|
# 前5根K线的价格趋势 (简单线性回归斜率)
|
|
if self.pre:
|
|
price_vals = [self.close]
|
|
temp = self.pre
|
|
count = 0
|
|
|
|
while temp and count < 5:
|
|
price_vals.append(temp.close)
|
|
count += 1
|
|
temp = temp.pre
|
|
|
|
if len(price_vals) > 2:
|
|
import numpy as np
|
|
y = np.array(price_vals)
|
|
x = np.arange(len(y))
|
|
|
|
# 简单线性回归
|
|
slope = np.polyfit(x, y, 1)[0]
|
|
|
|
# 归一化斜率
|
|
if abs(np.mean(y)) > 0: # 避免除以零
|
|
features['klc_trend_slope'] = slope / abs(np.mean(y))
|
|
else:
|
|
features['klc_trend_slope'] = 0
|
|
else:
|
|
features['klc_trend_slope'] = 0
|
|
else:
|
|
features['klc_trend_slope'] = 0
|
|
|
|
# ===== 2.5 其他衍生因子 =====
|
|
|
|
# K线组合形态
|
|
# 十字星 (实体非常小)
|
|
body_pct = abs(self.close - self.open) / (self.high - self.low) if (self.high - self.low) > 0 else 0
|
|
features['klc_is_doji'] = 1 if body_pct < 0.1 else 0 # 实体小于10%算十字星
|
|
|
|
# 锤子线/上吊线 (下影线长，上影线短，实体小)
|
|
if high_low_range > 0:
|
|
lower_shadow_pct = (min_oc - self.low) / high_low_range
|
|
upper_shadow_pct = (self.high - max_oc) / high_low_range
|
|
features['klc_is_hammer'] = 1 if (lower_shadow_pct > 0.6 and upper_shadow_pct < 0.1) else 0
|
|
else:
|
|
features['klc_is_hammer'] = 0
|
|
|
|
# 吞没形态
|
|
if self.pre:
|
|
prev_body_size = abs(self.pre.close - self.pre.open)
|
|
curr_body_size = abs(self.close - self.open)
|
|
|
|
# 看涨吞没
|
|
if (self.pre.close < self.pre.open # 前一根是阴线
|
|
and self.close > self.open # 当前是阳线
|
|
and self.open <= self.pre.close # 当前开盘低于前收盘
|
|
and self.close >= self.pre.open # 当前收盘高于前开盘
|
|
and curr_body_size > prev_body_size): # 当前实体大于前实体
|
|
features['klc_is_bullish_engulfing'] = 1
|
|
else:
|
|
features['klc_is_bullish_engulfing'] = 0
|
|
|
|
# 看跌吞没
|
|
if (self.pre.close > self.pre.open # 前一根是阳线
|
|
and self.close < self.open # 当前是阴线
|
|
and self.open >= self.pre.close # 当前开盘高于前收盘
|
|
and self.close <= self.pre.open # 当前收盘低于前开盘
|
|
and curr_body_size > prev_body_size): # 当前实体大于前实体
|
|
features['klc_is_bearish_engulfing'] = 1
|
|
else:
|
|
features['klc_is_bearish_engulfing'] = 0
|
|
else:
|
|
features['klc_is_bullish_engulfing'] = 0
|
|
features['klc_is_bearish_engulfing'] = 0
|
|
|
|
# 包含关系
|
|
if self.pre:
|
|
# 向上包含
|
|
if (self.high >= self.pre.high and self.low >= self.pre.low):
|
|
features['klc_is_up_inclusive'] = 1
|
|
else:
|
|
features['klc_is_up_inclusive'] = 0
|
|
|
|
# 向下包含
|
|
if (self.high <= self.pre.high and self.low <= self.pre.low):
|
|
features['klc_is_down_inclusive'] = 1
|
|
else:
|
|
features['klc_is_down_inclusive'] = 0
|
|
|
|
# 完全包含
|
|
if (self.high >= self.pre.high and self.low <= self.pre.low):
|
|
features['klc_is_full_inclusive'] = 1
|
|
else:
|
|
features['klc_is_full_inclusive'] = 0
|
|
|
|
# 被完全包含
|
|
if (self.high <= self.pre.high and self.low >= self.pre.low):
|
|
features['klc_is_inner_inclusive'] = 1
|
|
else:
|
|
features['klc_is_inner_inclusive'] = 0
|
|
else:
|
|
features['klc_is_up_inclusive'] = 0
|
|
features['klc_is_down_inclusive'] = 0
|
|
features['klc_is_full_inclusive'] = 0
|
|
features['klc_is_inner_inclusive'] = 0
|
|
|
|
# 从KLU获取其他特征
|
|
features.update(self.cal_klu_features())
|
|
|
|
return features |