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

refactor chanlun, tf_df

Browse Source
This commit is contained in:
jackyu66git
2025-10-31 20:34:27 +08:00
parent fe4193b153
commit 8bb3798a7c
7 changed files with 618 additions and 1650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TF_DF.py
+396 -73
View File
@@ -1,6 +1,6 @@
from datetime import timedelta
from pandas import DataFrame
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_SEG_DIR, Chan_ZS_DIR, Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_PRICE_TREND
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_SEG_DIR, Chan_ZS_DIR, Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_PRICE_TREND, Chan_KLU_PATTERN
from ChanKLU import ChanKLU
from ChanKLC import ChanKLC
from ChanBI import ChanBI
@@ -10,17 +10,16 @@ from ChanZS import ChanZS
from ChanBSP import ChanBSP
import talib.abstract as ta
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.dates import DateFormatter, date2num
import matplotlib.patches as patches
from technical.util import resample_to_interval
from decimal import Decimal
import xgboost as xgb
import numpy as np
from ChanMACD import ChanMACD
class TF_DF():
def __init__(self, df, interval, timeframe):
def __init__(self, df=None, interval=0, timeframe=None):
if df is not None:
self.init_TF_DF(df, interval, timeframe)
def init_TF_DF(self, df, interval, timeframe):
self.timeframe = timeframe
self.interval = interval
self.dataframe = resample_to_interval(df, interval)
@@ -31,12 +30,10 @@ class TF_DF():
self.zs_list = []
self.bsp_list = []
self.seg_list = []
self.init_TF_DF()
def init_TF_DF(self):
self.klu_list = self.cal_kl_data(self.dataframe)
self.klc_list = self.cal_klc_list(self.klu_list)
self.klc_list = self.get_klc_list(self.klu_list)
self.bi_list = self.cal_bi_list(self.klc_list)
self.seg_list = self.cal_seg_list(self.bi_list)
self.seg_list = self.get_seg_list(self.bi_list)
self.zs_list = self.cal_zs_list(self.bi_list, self.seg_list)
self.chanmacd = ChanMACD(self.klu_list)
self.klu_list = self.chanmacd.cal_macd_state()
@@ -97,45 +94,45 @@ class TF_DF():
df['macd'] = macd['macd']
df['macdsignal'] = macd['macdsignal']
df['macdhist'] = macd['macdhist']
df['ema5'] = self.cal_ema(df, 5)
df['ema10'] = self.cal_ema(df, 10)
df['ema24'] = self.cal_ema(df, 24)
df['ema26'] = self.cal_ema(df, 26)
df['ema52'] = self.cal_ema(df, 52)
df['ema5'] = ta.EMA(df, timeperiod=5)
df['ema10'] = ta.EMA(df, timeperiod=10)
df['ema24'] = ta.EMA(df, timeperiod=24)
df['ema52'] = ta.EMA(df, timeperiod=52)
df['rsi'] = ta.RSI(df, timeperiod=14)
df['volume_ratio'] = self.cal_volume_ratio(df)
return df
@staticmethod
def cal_ema(df, timeperiod):
"""
计算 EMA，优先使用 pandas ewm(adjust=False) 以贴近前端/TradingView 显示；
必要时回退到 TA-Lib（abstract）。
"""
try:
series = df['close'].astype(float) if isinstance(df, pd.DataFrame) else pd.Series(df).astype(float)
return series.ewm(span=int(timeperiod), adjust=False).mean()
except Exception:
try:
if isinstance(df, pd.DataFrame):
return ta.EMA(df, timeperiod=int(timeperiod))
except Exception:
pass
# 最后回退：返回同索引的 NaN 序列
if isinstance(df, pd.DataFrame) and 'close' in df:
return pd.Series(np.nan, index=df.index)
return pd.Series(dtype=float)
def get_klu_state(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
klu_state_list = []
klc_index = 0
for index in range(0, len(dataframe)):
if klc_index == len(klc_list):
klc_index = len(klc_list) - 1
klc = klc_list[klc_index]
if klc.end_klu and klc.end_klu.idx == index:
if klc.klc_fx_type == Chan_KLC_FX.TOP4:
klu_state_list.append("10")
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM4:
klu_state_list.append("-10")
else:
klu_state_list.append("00")
klc_index += 1
else:
klu_state_list.append("00")
return klu_state_list
def check_fx(self, klc):
if klc.pre and klc.next:
if klc.high > klc.pre.high and klc.high > klc.next.high:
if klc.macd > 0 and klc.signal > klc.macdhist:
klc.set_fx(Chan_FX_TYPE.TOP)
# print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.low and klc.low < klc.next.low:
if klc.macd < 0 and klc.signal < klc.macdhist:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
# print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0 and klc.macd > klc.macdhist:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0 and klc.macd < klc.macdhist:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def cal_volume_ratio(self, dataframe, window=10):
df = dataframe.copy()
@@ -155,6 +152,9 @@ class TF_DF():
if not klc_list:
return klc_list
last_trend = Chan_PRICE_TREND.UNKNOWN
# 趋势延续性：参考近 N 根已完成的KLC
lookback_n = 5
prev_klcs = []
for klc in klc_list:
price = getattr(klc, 'close', None)
ema24 = getattr(klc, 'ema24', None)
@@ -164,13 +164,14 @@ class TF_DF():
hist = getattr(klc, 'macdhist', 0) if getattr(klc, 'macdhist', None) is not None else 0
rsi = getattr(klc, 'rsi', None)
trend = Chan_PRICE_TREND.UNKNOWN
score = 0
try:
# 有效性
price_valid = price is not None and price != 0
ema24_valid = ema24 is not None and ema24 != 0
ema52_valid = ema52 is not None and ema52 != 0
# 多因子投票
score = 0
# 1) 均线结构 + 价位
if ema24_valid and ema52_valid:
score += 1 if ema24 > ema52 else -1
@@ -194,6 +195,76 @@ class TF_DF():
spread_now = ema24 - ema52
spread_pre = pre_ema24 - pre_ema52
score += 1 if spread_now >= spread_pre else -1
# 3.1) MACD柱体动量趋势：考虑 macdhist 的斜率与过零
pre_hist = getattr(pre, 'macdhist', None)
if pre_hist is not None and hist is not None:
# 柱体斜率：上升加分，下降减分
if hist > pre_hist:
score += 1
elif hist < pre_hist:
score -= 1
# 过零加权：负转正更偏多，正转负更偏空
if pre_hist < 0 and hist > 0:
score += 1
elif pre_hist > 0 and hist < 0:
score -= 1
# 3.2) EMA52 突破/跌破加权
if ema52_valid and price_valid and pre_close is not None and pre_ema52 not in (None, 0):
# 看多突破：从均线下方上破且动量配合
if pre_close <= pre_ema52 and price > ema52 and (hist is None or pre_hist is None or hist >= pre_hist):
score += 1
# 看空跌破：从均线上方下破且动量配合
if pre_close >= pre_ema52 and price < ema52 and (hist is None or pre_hist is None or hist <= pre_hist):
score -= 1
# 3.3) EMA52 支撑/阻力触碰（非强穿越）
if ema52_valid and price_valid:
low_v = getattr(klc, 'low', None)
high_v = getattr(klc, 'high', None)
if low_v is not None and high_v is not None and ema52 not in (None, 0):
# 触碰容差（相对EMA52的0.15%）
touch_tol = 0.0015
# 作为支撑：收盘在上，最低靠近EMA52
near_support_touch = (price > ema52) and (abs(low_v - ema52) / abs(ema52) <= touch_tol)
# 作为阻力：收盘在下，最高靠近EMA52
near_resistance_touch = (price < ema52) and (abs(high_v - ema52) / abs(ema52) <= touch_tol)
if near_support_touch:
# 若动量不弱，则更偏多
score += 1 if (hist is None or pre_hist is None or hist >= pre_hist) else 0
if near_resistance_touch:
# 若动量不强，则更偏空
score -= 1 if (hist is None or pre_hist is None or hist <= pre_hist) else 0
# 3.4) 多次对 EMA52 的"拒绝"配合 MACD 逆向：易形成压/支并反向
# 统计近窗口内的上/下拒绝次数：
# - 上拒绝：价格位于 EMA52 下方，最高触及/越过 EMA52 但收盘仍在下方
# - 下拒绝：价格位于 EMA52 上方，最低触及/跌破 EMA52 但收盘仍在上方
recent_up_rejects = 0
recent_down_rejects = 0
if ema52_valid:
window_rej = prev_klcs[-lookback_n:] if len(prev_klcs) > 0 else []
rej_tol = 0.0015
for wk in window_rej:
wk_close = getattr(wk, 'close', None)
wk_ema52 = getattr(wk, 'ema52', None)
wk_high = getattr(wk, 'high', None)
wk_low = getattr(wk, 'low', None)
if wk_close is None or wk_ema52 in (None, 0):
continue
# 上拒绝（阻力）：下方多次试图上破但未站上
if wk_close < wk_ema52 and wk_high is not None:
if wk_high >= wk_ema52 or abs(wk_high - wk_ema52) / abs(wk_ema52) <= rej_tol:
recent_up_rejects += 1
# 下拒绝（支撑）：上方多次试图下破但未跌破
if wk_close > wk_ema52 and wk_low is not None:
if wk_low <= wk_ema52 or abs(wk_low - wk_ema52) / abs(wk_ema52) <= rej_tol:
recent_down_rejects += 1
# 定义 MACD 的方向偏好
macd_bias_up = (macd >= signal) and (hist is None or pre_hist is None or hist >= pre_hist)
macd_bias_down = (macd <= signal) and (hist is None or pre_hist is None or hist <= pre_hist)
# 若多次上拒绝且 MACD 偏空，则更偏向下行；若多次下拒绝且 MACD 偏多，则更偏向上行
if recent_up_rejects >= 2 and macd_bias_down:
score -= 2
if recent_down_rejects >= 2 and macd_bias_up:
score += 2
# 4) RSI 辅助
if rsi is not None:
if rsi >= 55:
@@ -231,17 +302,121 @@ class TF_DF():
near_macd = abs(macd - signal) <= (abs(price) * 0.00005 if price_valid else 0)
near_flat = near_ema52 and near_macd
# 7) 动态阈值 + 趋势记忆（更强粘滞：趋势中容忍小幅反分）
# 引入过去 N 根KLC 的趋势延续性来动态调整翻转阈值，并结合 EMA52 支撑/阻力触碰强化门槛
force_flip_down = False
force_flip_up = False
if near_flat:
trend = Chan_PRICE_TREND.FLAT
else:
if last_trend == Chan_PRICE_TREND.UP:
# 仅当出现明显反向才翻转，否则维持UP
if score <= -2:
# 计算过去窗口的趋势一致性
window = prev_klcs[-lookback_n:] if len(prev_klcs) > 0 else []
persist_up = 0
persist_down = 0
for wk in window:
if getattr(wk, 'trend', None) == Chan_PRICE_TREND.UP:
persist_up += 1
elif getattr(wk, 'trend', None) == Chan_PRICE_TREND.DOWN:
persist_down += 1
persist_ratio_up = (persist_up / len(window)) if len(window) > 0 else 0
persist_ratio_down = (persist_down / len(window)) if len(window) > 0 else 0
# 基准阈值
down_flip_threshold = -2
up_flip_threshold = 2
# 若最近多为UP，则从UP翻转需更强反向信号；同理对DOWN
if last_trend == Chan_PRICE_TREND.UP and persist_ratio_up >= 0.6:
down_flip_threshold = -3
elif last_trend == Chan_PRICE_TREND.DOWN and persist_ratio_down >= 0.6:
up_flip_threshold = 3
# EMA52 触碰强化门槛：UP时若出现支撑触碰，下翻更难；DOWN时若出现阻力触碰，上翻更难
if ema52_valid and price_valid:
low_v = getattr(klc, 'low', None)
high_v = getattr(klc, 'high', None)
if low_v is not None and high_v is not None and ema52 not in (None, 0):
touch_tol = 0.0015
near_support_touch = (price > ema52) and (abs(low_v - ema52) / abs(ema52) <= touch_tol)
near_resistance_touch = (price < ema52) and (abs(high_v - ema52) / abs(ema52) <= touch_tol)
if last_trend == Chan_PRICE_TREND.UP and near_support_touch:
# 强化维持UP：进一步降低向下翻转阈值
down_flip_threshold = min(down_flip_threshold - 1, -3)
if last_trend == Chan_PRICE_TREND.DOWN and near_resistance_touch:
# 强化维持DOWN：进一步提高向上翻转阈值
up_flip_threshold = max(up_flip_threshold + 1, 3)
# 7.1) 复合拐头信号：MACD/Signal 同向拐头 + hist 连续减弱 + 多次未能越过 EMA52
pre_macd = getattr(pre, 'macd', None) if pre else None
pre_signal = getattr(pre, 'signal', None) if pre else None
macd_slope = (macd - pre_macd) if (pre_macd is not None and macd is not None) else 0
signal_slope = (signal - pre_signal) if (pre_signal is not None and signal is not None) else 0
# hist 连续减弱（绝对值缩小）
hist_seq = []
for wk in prev_klcs[-2:]:
val = getattr(wk, 'macdhist', None)
if val is not None:
hist_seq.append(val)
if hist is not None:
hist_seq.append(hist)
weaken_steps = 0
for i in range(1, len(hist_seq)):
if abs(hist_seq[i]) < abs(hist_seq[i-1]):
weaken_steps += 1
# 近窗口对 EMA52 的"未能站上/跌破"统计（放宽窗口与条件）
window_ema = prev_klcs[-4:] if len(prev_klcs) > 0 else []
no_up_break = False
no_down_break = False
if ema52_valid:
# 未能有效上破：最近若干根收盘大多数不在 EMA52 上方，且高点多次触及/接近
cnt_touch_up = 0
cnt_close_above = 0
for wk in window_ema:
wk_close = getattr(wk, 'close', None)
wk_high = getattr(wk, 'high', None)
wk_ema = getattr(wk, 'ema52', None)
if wk_close is not None and wk_ema not in (None, 0):
if wk_close > wk_ema:
cnt_close_above += 1
if wk_high is not None and (wk_high >= wk_ema or abs(wk_high - wk_ema) / abs(wk_ema) <= 0.0015):
cnt_touch_up += 1
no_up_break = (cnt_close_above <= 1 and cnt_touch_up >= 1 and price <= ema52)
# 未能有效下破：最近若干根收盘大多数不在 EMA52 下方，且低点多次触及/接近
cnt_touch_down = 0
cnt_close_below = 0
for wk in window_ema:
wk_close = getattr(wk, 'close', None)
wk_low = getattr(wk, 'low', None)
wk_ema = getattr(wk, 'ema52', None)
if wk_close is not None and wk_ema not in (None, 0):
if wk_close < wk_ema:
cnt_close_below += 1
if wk_low is not None and (wk_low <= wk_ema or abs(wk_low - wk_ema) / abs(wk_ema) <= 0.0015):
cnt_touch_down += 1
no_down_break = (cnt_close_below <= 1 and cnt_touch_down >= 1 and price >= ema52)
# 若当前为UP趋势，出现明显拐头+hist减弱+未能上破EMA52，则加速看空
if last_trend == Chan_PRICE_TREND.UP and macd_slope < 0 and signal_slope < 0 and weaken_steps >= 1 and no_up_break and macd_bias_down:
score -= 3
down_flip_threshold = max(down_flip_threshold, 0)
force_flip_down = True
# 若当前为DOWN趋势，出现明显拐头+hist减弱+未能下破EMA52，则加速看多
if last_trend == Chan_PRICE_TREND.DOWN and macd_slope > 0 and signal_slope > 0 and weaken_steps >= 1 and no_down_break and macd_bias_up:
score += 3
up_flip_threshold = min(up_flip_threshold, 0)
force_flip_up = True
# 多次对 EMA52 的拒绝配合 MACD 逆向：加速反向翻转（降低相反方向阈值）
if recent_up_rejects >= 2 and macd_bias_down:
# 从 UP 向 DOWN 的翻转更容易
down_flip_threshold = max(down_flip_threshold, -1)
if recent_down_rejects >= 2 and macd_bias_up:
# 从 DOWN 向 UP 的翻转更容易
up_flip_threshold = min(up_flip_threshold, 1)
if force_flip_down:
trend = Chan_PRICE_TREND.DOWN
elif force_flip_up:
trend = Chan_PRICE_TREND.UP
elif last_trend == Chan_PRICE_TREND.UP:
if score <= down_flip_threshold:
trend = Chan_PRICE_TREND.DOWN
else:
trend = Chan_PRICE_TREND.UP
elif last_trend == Chan_PRICE_TREND.DOWN:
if score >= 2:
if score >= up_flip_threshold:
trend = Chan_PRICE_TREND.UP
else:
trend = Chan_PRICE_TREND.DOWN
@@ -256,14 +431,19 @@ class TF_DF():
except Exception:
trend = Chan_PRICE_TREND.UNKNOWN
# 写回趋势
if klc.end_time is None:
trend = Chan_PRICE_TREND.FLAT
if hasattr(klc, 'set_trend'):
klc.set_trend(trend)
else:
setattr(klc, 'trend', trend)
last_trend = trend
# 更新滑窗：仅向后看
prev_klcs.append(klc)
price_diff = klc.close - klc.pre.close if klc.pre else 0
#print(klc.start_time, klc.end_time, klc.close, klc.ema24, klc.ema52, klc.macd, klc.signal, klc.macdhist, klc.trend, price_diff)
#print(klc.start_time, klc.end_time, klc.trend, price_diff)
#if klc.index > len(klc_list) - 10:
#print(klc.start_time, klc.end_time, klc.close, klc.ema24, klc.ema52, klc.macd, klc.signal, klc.macdhist, klc.trend, price_diff, score)
#print(klc.start_time, klc.end_time, klc.trend, price_diff, score)
return klc_list
def cal_kl_data(self, dataframe:DataFrame):
fields = "time,open,high,low,close,volume"
@@ -300,8 +480,46 @@ class TF_DF():
if 'macd' in item:
klu.set_indicators(item)
return klu_list
def cal_klc_list(self, klu_list):
def get_bi_list(self, dataframe):
bi_list = self.cal_bi_list(self.get_klc_list(dataframe))
return bi_list
def get_kl_data(self, dataframe:DataFrame):
fields = "time,open,high,low,close,volume"
klu_list = []
last_klu = None
for i in range(0, len(dataframe)):
item = dataframe.iloc[i]
date = item['date']
o = item['open']
h = item['high']
l = item['low']
c = item['close']
v = item['volume']
#time_obj = date.fromtimestamp(date)
#date = date + timedelta(hours=8)
time_str = date.strftime('%Y-%m-%d %H:%M:%S')
item_data = [
time_str,
o,
h,
l,
c,
v
]
#klu = KLU(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)))
klu = ChanKLU(time_str, o, h, l, c, v)
#print(klu.time, klu.open, klu.high, klu.low, klu.close, klu.volume)
klu.set_idx(i)
klu_list.append(klu)
if last_klu:
last_klu.set_next(klu)
klu.set_pre(last_klu)
last_klu = klu
if 'macd' in item:
klu.set_indicators(item)
return klu_list
def get_klc_list(self, dataframe):
klu_list = self.get_klu_list(dataframe)
klc_list = []
last_klu = None
macd = ChanMACD(klu_list)
@@ -309,19 +527,33 @@ class TF_DF():
for klu in klu_list:
if len(klc_list) > 0:
last_klc = klc_list[-1]
included = last_klc.check_klu_included(klu)
if not included:
if klu.exception:
ddir = Chan_KLINE_DIR.DOWN
if last_klc.high < klu.high:
ddir = Chan_KLINE_DIR.UP
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
klc.high = klu.close if klu.close > klu.open else klu.open
klc.low = klu.open if klu.close > klu.open else klu.close
klc_list.append(klc)
last_klc.set_next(klc)
klc.set_pre(last_klc)
last_klc.set_end_klu(last_klu)
klc.set_pre_fx()
#print(klu.time, klu.high, klu.low, klu.close, klu.open, klu.exception)
else:
last_klc.add_klu(klu)
included = last_klc.check_klu_included(klu)
if not included:
ddir = Chan_KLINE_DIR.DOWN
if last_klc.high < klu.high:
ddir = Chan_KLINE_DIR.UP
klc = ChanKLC(klu, index=len(klc_list), ddir=ddir)
klc_list.append(klc)
last_klc.set_next(klc)
klc.set_pre(last_klc)
last_klc.set_end_klu(last_klu)
klc.set_pre_fx()
else:
last_klc.add_klu(klu)
else:
ddir = Chan_KLINE_DIR.UP
if klu.open > klu.close:
@@ -332,7 +564,7 @@ class TF_DF():
klc_list = self.cal_trend(klc_list)
return klc_list
def cal_seg_list(self, bi_list):
def get_seg_list(self, bi_list):
seg_list = []
up_bi_list = []
down_bi_list = []
@@ -371,7 +603,7 @@ class TF_DF():
if last_down_sbi.has_fx_gap:
look_for_bottom = True
last_seg.pre_set_end_bi(bi_list[last_down_sbi.start_bi.index - 1])
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN)
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN, bi)
seg_list.append(seg)
last_seg.set_next(seg)
seg.set_pre(last_seg)
@@ -397,7 +629,7 @@ class TF_DF():
#print(bi.start_time, look_for_top, "UP 3")
else:
last_seg.set_end_bi(bi_list[last_down_sbi.start_bi.index - 1], bi)
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN)
seg = ChanSEG(last_down_sbi.start_bi, len(seg_list), Chan_SEG_DIR.DOWN, bi)
seg_list.append(seg)
last_seg.set_next(seg)
seg.set_pre(last_seg)
@@ -465,7 +697,7 @@ class TF_DF():
if last_up_sbi.has_fx_gap:
look_for_top = True
last_seg.pre_set_end_bi(bi_list[last_up_sbi.start_bi.index - 1])
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP)
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP, bi)
seg_list.append(seg)
last_seg.set_next(seg)
seg.set_pre(last_seg)
@@ -491,7 +723,7 @@ class TF_DF():
#print(bi.start_time, look_for_top, "DOWN 3")
else:
last_seg.set_end_bi(bi_list[last_up_sbi.start_bi.index - 1], bi)
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP)
seg = ChanSEG(last_up_sbi.start_bi, len(seg_list), Chan_SEG_DIR.UP, bi)
#print(last_up_sbi.start_bi.start_time)
last_seg.set_next(seg)
seg.set_pre(last_seg)
@@ -539,14 +771,14 @@ class TF_DF():
else:
if bi.check_overlap():
if bi.dir == Chan_BI_DIR.UP:
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.UP)
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.UP, bi)
last_up_bi = bi
last_up_sbi = ChanSBI(bi, len(up_sbi_list), bi.dir)
seg_list.append(seg)
last_seg = seg
#print(bi.start_time, 'Create first UP SEG')
else:
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.DOWN)
seg = ChanSEG(bi, len(seg_list), Chan_SEG_DIR.DOWN, bi)
last_down_bi = bi
last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
seg_list.append(seg)
@@ -573,7 +805,7 @@ class TF_DF():
# The confirmed
print("Last UP seg is broken, create a new seg. 1")
seg.pre_set_end_bi(bi_list[i])
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.DOWN)
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.DOWN, bi)
seg_list.append(seg)
last_seg = seg_list[-2]
if len(last_seg.bi_list) > 3:
@@ -585,7 +817,7 @@ class TF_DF():
if bi_list[i].low < last_seg_peak:
print("Last DOWN seg is broken, create a new seg. 1")
seg.pre_set_end_bi(bi_list[i])
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.UP)
seg = ChanSEG(bi_list[i+1], len(seg_list), Chan_SEG_DIR.UP, bi)
seg_list.append(seg)
last_seg = seg_list[-2]
if len(last_seg.bi_list) > 3:
@@ -603,12 +835,13 @@ class TF_DF():
if bi_list[i].high > last_seg_peak:
print("Last seg is broken, create a new seg. 2")
last_seg.pre_set_end_bi(bi_list[i-1])
seg = ChanSEG(bi_list[i], len(seg_list), Chan_SEG_DIR.UP)
seg = ChanSEG(bi_list[i], len(seg_list), Chan_SEG_DIR.UP, bi)
seg_list.append(seg)
last_seg = seg
last_seg_bi = bi_list[i]
break
"""
self.cal_bi_zs(seg_list)
return seg_list
def cal_bi_list(self, klc_list):
@@ -617,7 +850,14 @@ class TF_DF():
last_bottom = None
for klc in klc_list:
fx = self.check_fx(klc)
if fx == Chan_FX_TYPE.TOP:
if last_bottom:
if self.check_top_fx(last_bottom, klc) == False:
fx = Chan_FX_TYPE.UNKNOWN
if fx == Chan_FX_TYPE.BOTTOM:
if last_top:
if self.check_bottom_fx(last_top, klc) == False:
fx = Chan_FX_TYPE.UNKNOWN
# Do nothing
if fx == Chan_FX_TYPE.UNKNOWN:
continue
@@ -665,6 +905,7 @@ class TF_DF():
#klc.set_state("20")
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
#klc.cal_invisible()
#klc.set_klc_fx_type(Chan_KLC_FX.TOP3)
#print(klc.start_time, klc.fx, "二类卖点Sell 1")
else:
@@ -789,6 +1030,7 @@ class TF_DF():
#klc.set_state("-20")
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
#klc.cal_invisible()
#klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3)
#print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1")
#print(klc.start_time, klc.fx, "二类买点Buy 1")
@@ -917,11 +1159,25 @@ class TF_DF():
#for index in range(0, 10):
#print(bi_list[index].start_time, bi_list[index].start_klc.start_time, bi_list[index].dir)
return bi_list
def check_top_fx(self, last_bottom, klc):
if last_bottom.high > klc.pre.low or last_bottom.high > klc.next.low:
return False
return True
def check_bottom_fx(self, last_top, klc):
if last_top.low < klc.pre.high or last_top.low < klc.next.high:
return False
return True
def get_decimal(self, value):
return Decimal("{:.2f}".format(value))
def cal_bi_zs(self, seg_list):
bi_zs_list = []
for seg in seg_list:
zs_list = seg.cal_bi_zs()
if len(zs_list) > 0:
bi_zs_list.append(zs_list)
return bi_zs_list
def cal_zs_list(self, bi_list, seg_list):
def get_zs_list(self, bi_list, seg_list):
zs_list = []
bsp_list = []
if len(seg_list) > 3:
@@ -940,7 +1196,6 @@ class TF_DF():
zd = max(seg.low, seg.next.low, seg.next.next.low)
gg = max(seg.high, seg.next.high, seg.next.next.high)
dd = min(seg.low, seg.next.low, seg.next.next.low)
ddir = Chan_ZS_DIR.UP
ddir = None
if last_zs:
if zg < last_zs.zd:
@@ -981,7 +1236,7 @@ class TF_DF():
# SEG is not in ZS
if seg.is_sure:
if ((seg.low > last_zs.zg and seg.high > last_zs.zg) or (seg.high < last_zs.zd and seg.low < last_zs.zd)):
last_zs.set_end_klc(last_zs.last_bi_in.end_klc, seg.sure_time, bi_out_count, seg)
last_zs.set_end_klc(seg.pre.end_bi.end_klc, seg.sure_time, bi_out_count, seg.pre)
bi_out_count = 0
#print(seg.start_bi.start_klc.start_time)
first_bi_out = None
@@ -1095,4 +1350,72 @@ class TF_DF():
bsp = ChanBSP(bi, len(bsp_list), Chan_BSP_TYPE.T3E, Chan_BSP_DIR.BUY if bi.dir == Chan_BI_DIR.DOWN else Chan_BSP_DIR.SELL, bi.sure_time, zs_count, zs, seg)
bsp_list.append(bsp)
#self.print_zs(zs_list)
return zs_list
return zs_list
def get_klu_list(self, dataframe):
klu_list = self.get_kl_data(dataframe)
return self.cal_klu_pattern(klu_list)
def cal_klu_pattern(self, klu_list):
"""
计算裸K的pattern - 只识别反转形态
"""
if not klu_list or len(klu_list) < 3:
return klu_list
for i, klu in enumerate(klu_list):
# 单根K线反转模式识别
self._detect_single_reversal_pattern(klu)
if klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
print(klu.time, klu.pattern)
return klu_list
def _detect_single_reversal_pattern(self, klu):
"""检测单根K线反转模式"""
body = abs(klu.close - klu.open)
upper_shadow = klu.high - max(klu.close, klu.open)
lower_shadow = min(klu.close, klu.open) - klu.low
total_range = klu.high - klu.low
# 避免除零
if total_range == 0:
return
body_ratio = body / total_range
upper_ratio = upper_shadow / total_range
lower_ratio = lower_shadow / total_range
# 锤子线/上吊线 - 反转信号
if lower_ratio / body_ratio >= 2:
# 锤子线：底部反转，需要前面一段
if klu.close > klu.open and klu.pre and klu.pre.close < klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.HAMMER) # 底部反转
# 上吊线：顶部反转，需要前一根是上涨趋势
elif klu.close < klu.open and klu.pre and klu.pre.close > klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.HANGING_MAN) # 顶部反转
# 倒锤子线/射击之星 - 反转信号
elif upper_ratio / body_ratio >= 2:
# 倒锤子线：底部反转，需要前一根是下跌趋势
if klu.close > klu.open and klu.pre and klu.pre.close < klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.INVERTED_HAMMER) # 底部反转
# 射击之星：顶部反转，需要前一根是上涨趋势
elif klu.close < klu.open and klu.pre and klu.pre.close > klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.SHOOTING_STAR) # 顶部反转
# 十字星 - 反转信号
elif body_ratio <= 0.1:
if upper_ratio > 0.4 and lower_ratio > 0.4:
klu.set_pattern(Chan_KLU_PATTERN.LONG_LEGGED_DOJI) # 强烈反转信号
elif upper_ratio > 0.4 and lower_ratio <= 0.1:
# 墓碑十字星：顶部反转，需要前一根是上涨趋势
if klu.pre and klu.pre.close > klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.GRAVESTONE_DOJI) # 顶部反转
elif lower_ratio > 0.4 and upper_ratio <= 0.1:
# 蜻蜓十字星：底部反转，需要前一根是下跌趋势
if klu.pre and klu.pre.close < klu.pre.open:
klu.set_pattern(Chan_KLU_PATTERN.DRAGONFLY_DOJI) # 底部反转
else:
klu.set_pattern(Chan_KLU_PATTERN.DOJI) # 一般反转信号
def get_decimal(self, value):
return Decimal("{:.2f}".format(value))