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

add macd判断背离，使用分型和隐形形态实现第二类买卖点

Browse Source
This commit is contained in:
jackyu66git
2025-08-18 01:14:36 +08:00
parent 45595fe0f8
commit b15ba06a00
12 changed files with 2422 additions and 816 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanEnum.py
+27 -8
View File
@@ -67,26 +67,45 @@ class Chan_KLC_FX(Enum):
TOP2 = auto()
TOP3 = auto()
TOP4 = auto()
TOP5 = auto()
BOTTOM1 = auto()
BOTTOM2 = auto()
BOTTOM3 = auto()
BOTTOM4 = auto()
BOTTOM5 = auto()
UNKNOWN = auto()
# 统一的MACD状态枚举，包含所有可能的状态
class Chan_MACD_STATE(Enum):
GW = auto()
GWK = auto()
UP = auto()
DOWN = auto()
CROSS0 = auto()
UNKNOWN = auto()
START = auto()
NEAR0 = auto()
"""MACD状态枚举 - 包含所有可能的状态"""
# 穿越状态
CROSS0_UP = auto() # 向上穿越零轴
CROSS0_DOWN = auto() # 向下穿越零轴
# 趋势状态
NEAR0 = auto() # 接近零轴
# 位置状态
HIGH = auto() # 高位：MACD黄白线离开零轴到高点，能量柱最大开始减弱
HIGH_EMPTY = auto() # 高位空：MACD黄白线处于高位，能量柱衰减，与黄白线形成空间夹角
RETURN_ZERO = auto() # 归零轴：能量柱呈现一根比一根短的排列方式
UP = auto() # 归零轴后的上涨
DOWN = auto() # 归零轴后的下跌
# 基础状态
UNKNOWN = auto() # 未知
START = auto() # 开始
class Chan_MACDSEG_DIR(Enum):
ABOVE = auto()
UNDER = auto()
class Chan_MACDUNITTF_TYPE(Enum):
START = auto()
CROSS0 = auto()
NEAR0 = auto()
class Chan_MACDHISTSET_DIR(Enum):
ABOVE = auto()
UNDER = auto()
class Chan_MACDUNITTF_DIR(Enum):
ABOVE = auto()
UNDER = auto()
class Chan_MACDHIST_STATE(Enum):
UP = auto()
DOWN = auto()
ChanKLC.py
+10 -5
View File
@@ -30,7 +30,7 @@ class ChanKLC():
self.klc_fx_type = Chan_KLC_FX.UNKNOWN
self.rsi = klu.rsi
self.volume_ratio = klu.volume_ratio
self.macdhist = 0
self.macdhist = klu.macdhist
self.body = klu.body
self.upper_shadow = klu.upper_shadow
self.lower_shadow = klu.lower_shadow
@@ -65,25 +65,30 @@ class ChanKLC():
self.cal_bb_out()
if self.pre:
self.pre.cal_bb_out()
#self.bb_out = True
def cal_bb_out(self):
for klu in self.klus:
if self.high >= klu.bbup302 and klu.bbup302 > 0 and (self.klc_fx_type == Chan_KLC_FX.TOP1 or self.klc_fx_type == Chan_KLC_FX.TOP2):
self.bb_out = True
#print(self.end_time, self.high, klu.bbup302, self.klc_fx_type)
if self.high >= klu.bbup30 and klu.bbup30 > 0 and self.next and (self.next.macd - self.macd) < 0:
self.klc_fx_type = Chan_KLC_FX.TOP4
self.klc_fx_type = Chan_KLC_FX.TOP4
if self.low <= klu.bblow302 and klu.bblow302 > 0 and (self.klc_fx_type == Chan_KLC_FX.BOTTOM1 or self.klc_fx_type == Chan_KLC_FX.BOTTOM2):
self.bb_out = True
if self.low <= klu.bblow30 and klu.bblow30 > 0 and self.next and (self.macd - self.next.macd) < 0:
self.klc_fx_type = Chan_KLC_FX.BOTTOM4
if self.fx ==Chan_FX_TYPE.TOP:
self.klc_fx_type = Chan_KLC_FX.BOTTOM4
if self.fx == Chan_FX_TYPE.TOP:
#print(self.end_time, self.fx, self.macd, self.macdhist, len(self.klus))
if self.macd > 0:
self.bb_out = True
if self.macdhist < 0:
self.klc_fx_type = Chan_KLC_FX.TOP5
else:
if self.fx == Chan_FX_TYPE.BOTTOM:
if self.macd < 0:
self.bb_out = True
if self.macdhist > 0:
self.klc_fx_type = Chan_KLC_FX.BOTTOM5
#self.bb_out = True
def cal_macd_state(self, dir):
macd_state = 0
return macd_state
ChanKLU.py
+743 -673
View File
File diff suppressed because it is too large Load Diff
ChanMACD.py
+77 -52
View File
@@ -1,5 +1,5 @@
from ChanKLU import ChanKLU
from ChanEnum import Chan_MACD_STATE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR
from ChanEnum import Chan_MACD_STATE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR, Chan_MACDUNITTF_DIR, Chan_MACDUNITTF_TYPE
from ChanMACDSeg import ChanMACDSeg
from ChanMACDUnitTF import ChanMACDUnitTF
from ChanMACDHistSet import ChanMACDHistSet
@@ -10,6 +10,12 @@ class ChanMACD():
self.seg_list = []
self.unittf_list = []
self.histset_list = []
# 状态标记列表
self.high_position_list = [] # 高位列表
self.high_empty_list = [] # 高位空列表
self.return_zero_list = [] # 归零轴列表
self.cross0_up_list = [] # 向上穿越零轴列表
self.cross0_down_list = [] # 向下穿越零轴列表
self.cal_macd()
def cal_macd(self):
last_seg = None
@@ -17,68 +23,98 @@ class ChanMACD():
last_histset = None
last_klu = None
for klu in self.klu_list:
for klu in self.klu_list:
klu.cal_macd_state()
# 1) 只有当 MACD 已可用(非 UNKNOWN)时，才开始初始化段/单元
if last_seg is None:
if klu.macd_state != Chan_MACD_STATE.UNKNOWN:
# 初始化首段与首单元
seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.macd >= 0 else Chan_MACDSEG_DIR.UNDER
# 初始化首个直方图集合（根据当前柱体正负）
if klu.macdhist >= 0:
histset = ChanMACDHistSet(klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
else:
histset = ChanMACDHistSet(klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
self.histset_list.append(histset)
last_histset = histset
# 初始化首段
seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.signal >= 0 else Chan_MACDSEG_DIR.UNDER
seg = ChanMACDSeg(klu.time, klu, None, seg_dir)
self.seg_list.append(seg)
last_seg = seg
unittf = ChanMACDUnitTF(klu.time, klu, None, None)
self.unittf_list.append(unittf)
last_unittf = unittf
# 初始化首个直方图集合（根据当前柱体正负）
if klu.macdhist >= 0:
histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.ABOVE)
else:
histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.UNDER)
self.histset_list.append(histset)
last_histset = histset
last_seg.add_histset(histset)
# 初始化首个UnitTF
unittf_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
unittf = ChanMACDUnitTF(klu.time, klu, None, unittf_dir, Chan_MACDUNITTF_TYPE.START)
self.unittf_list.append(unittf)
unittf.add_histset(histset)
last_unittf = unittf
last_unittf.add_histset(histset)
last_seg.add_unittf(unittf)
# 未就绪则继续等下一根；已就绪亦已完成首个结构初始化，继续下一根
last_klu = klu
continue
# 2) 过零切段/切单元
if klu.macd_state == Chan_MACD_STATE.CROSS0:
# 2) 过零切段（使用KLU中的穿越状态）
if (klu.macd_state == Chan_MACD_STATE.CROSS0_UP or
klu.macd_state == Chan_MACD_STATE.CROSS0_DOWN):
# 结束旧 unittf
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
# 新的单位时间周期
new_dir = Chan_MACDUNITTF_DIR.UNDER if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.ABOVE else Chan_MACDUNITTF_DIR.ABOVE
unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.CROSS0)
self.unittf_list.append(unittf)
unittf.add_histset(last_histset)
last_unittf.set_next(unittf)
last_unittf = unittf
# 收尾旧段
last_seg.end_klu = last_klu
last_seg.end_time = last_klu.time
last_seg.set_end_klu(last_klu)
# 新段方向取反
new_dir = Chan_MACDSEG_DIR.UNDER if last_seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else Chan_MACDSEG_DIR.ABOVE
seg = ChanMACDSeg(klu.time, klu, last_seg, new_dir)
self.seg_list.append(seg)
last_seg.set_next(seg)
last_seg = seg
# 切换 unit tf（以当前histset收尾为边界）
unittf = ChanMACDUnitTF(klu.time, klu, last_histset, None)
self.unittf_list.append(unittf)
if last_unittf:
last_unittf.end_klu = last_klu
last_unittf.end_time = last_klu.time
last_unittf.set_next(unittf)
last_unittf = unittf
last_seg.add_histset(last_histset)
last_seg.add_unittf(unittf)
else:
# 4) UnitTF 状态机：用黄线Signal的归零轴
if last_klu.macd_state == Chan_MACD_STATE.NEAR0 and last_unittf.near0_count > 1:
if klu.macd_state == Chan_MACD_STATE.UP:
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
new_dir = Chan_MACDUNITTF_DIR.ABOVE if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.UNDER else Chan_MACDUNITTF_DIR.UNDER
unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0)
self.unittf_list.append(unittf)
unittf.add_histset(last_histset)
last_unittf.set_next(unittf)
last_unittf = unittf
last_seg.add_unittf(unittf)
elif klu.macd_state == Chan_MACD_STATE.DOWN:
last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
new_dir = Chan_MACDUNITTF_DIR.UNDER if last_unittf.unittf_dir == Chan_MACDUNITTF_DIR.ABOVE else Chan_MACDUNITTF_DIR.ABOVE
unittf = ChanMACDUnitTF(klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0)
self.unittf_list.append(unittf)
unittf.add_histset(last_histset)
last_unittf.set_next(unittf)
last_unittf = unittf
last_seg.add_unittf(unittf)
else:
last_unittf.add_klu(klu)
last_seg.add_klu(klu)
last_histset.add_klu(klu)
else:
last_unittf.add_klu(klu)
last_seg.add_klu(klu)
last_histset.add_klu(klu)
# 3) 直方图集合（基于当前 unittf）
if klu.macdhist >= 0:
if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
last_histset.add_klu(klu)
else:
# 结束旧 histset（以前一根结束更合理）
if last_histset:
end_klu = getattr(klu, 'pre', None) or klu
last_histset.end_klu = end_klu
last_histset.end_time = end_klu.time
histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.ABOVE)
if last_histset and last_klu:
last_histset.set_end_klu(last_klu)
histset = ChanMACDHistSet(klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
self.histset_list.append(histset)
if last_histset:
last_histset.set_next(histset)
@@ -91,11 +127,10 @@ class ChanMACD():
if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.UNDER:
last_histset.add_klu(klu)
else:
if last_histset:
end_klu = getattr(klu, 'pre', None) or klu
last_histset.end_klu = end_klu
last_histset.end_time = end_klu.time
histset = ChanMACDHistSet(klu.time, klu, last_unittf, Chan_MACDHISTSET_DIR.UNDER)
# 结束旧 histset（以前一根结束更合理）
if last_histset and last_klu:
last_histset.set_end_klu(last_klu)
histset = ChanMACDHistSet(klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
self.histset_list.append(histset)
if last_histset:
last_histset.set_next(histset)
@@ -104,14 +139,4 @@ class ChanMACD():
last_seg.add_histset(histset)
if last_unittf:
last_unittf.add_histset(histset)
last_klu = klu
# 循环结束后，收尾当前打开的结构
if last_seg and getattr(last_seg, 'end_klu', None) is None:
last_seg.end_klu = last_klu
last_seg.end_time = last_klu.time
if last_unittf and getattr(last_unittf, 'end_klu', None) is None:
last_unittf.end_klu = last_klu
last_unittf.end_time = last_klu.time
if last_histset and getattr(last_histset, 'end_klu', None) is None:
last_histset.end_klu = last_klu
last_histset.end_time = last_klu.time
last_klu = klu
ChanMACDHistSet.py
+8 -3
View File
@@ -1,15 +1,20 @@
class ChanMACDHistSet():
def __init__(self, start_time, start_klu, hist_set_dir, pre_histset):
def __init__(self, start_time, start_klu, pre_histset, dir):
self.start_time = start_time
self.end_time = None
self.klu_list = []
self.klu_list.append(start_klu)
self.ref_klu = None
self.hist_set_dir = hist_set_dir
self.histset_dir = dir
self.next = None
self.pre = pre_histset
def set_next(self, next_histset):
self.next = next_histset
def add_klu(self, klu):
self.klu_list.append(klu)
klu.set_histset(self)
klu.set_histset(self)
def set_end_klu(self, end_klu):
self.end_klu = end_klu
self.end_time = end_klu.time
ChanMACDSeg.py
+16 -4
View File
@@ -1,4 +1,4 @@
from ChanEnum import Chan_MACDSEG_DIR
class ChanMACDSeg():
@@ -14,14 +14,26 @@ class ChanMACDSeg():
self.seg_dir = seg_dir
self.pre = pre_seg
self.next = None
self.peak_klu = start_klu
def set_next(self, next_seg):
self.next = next_seg
def add_klu(self, klu):
self.klu_list.append(klu)
klu.set_seg(self)
if klu:
self.klu_list.append(klu)
klu.set_seg(self)
if self.seg_dir == Chan_MACDSEG_DIR.ABOVE:
if klu.high > self.peak_klu.high:
self.peak_klu = klu
else:
if klu.low < self.peak_klu.low:
self.peak_klu = klu
def add_unittf(self, unittf):
self.unittf_list.append(unittf)
unittf.set_next(self)
def add_histset(self, histset):
self.hist_set.append(histset)
histset.set_next(self)
histset.set_next(self)
def set_end_klu(self, end_klu):
self.add_klu(end_klu)
self.end_klu = end_klu
self.end_time = end_klu.time
ChanMACDUnitTF.py
+20 -6
View File
@@ -1,8 +1,8 @@
from ChanEnum import Chan_MACD_STATE
class ChanMACDUnitTF():
def __init__(self, start_time, start_klu, start_histset, pre_unittf, dir):
def __init__(self, start_time, start_klu, pre_unittf, dir, start_type):
self.start_time = start_time
self.end_time = None
self.start_klu = start_klu
@@ -10,15 +10,29 @@ class ChanMACDUnitTF():
self.klu_list = []
self.klu_list.append(start_klu)
self.histset_list = []
self.histset_list.append(start_histset)
self.next = None
self.pre = pre_unittf
self.uinttf_dir = dir
self.unittf_dir = dir
self.start_type = start_type
self.end_type = None
self.peak_hist = start_klu.macdhist
self.near0_count = 1
def set_next(self, next_unittf):
self.next = next_unittf
def add_histset(self, histset):
self.histset_list.append(histset)
histset.set_next(self)
def add_klu(self, klu):
self.klu_list.append(klu)
klu.set_unittf(self)
if klu:
self.klu_list.append(klu)
klu.set_unittf(self)
if klu.macdhist > self.peak_hist:
self.peak_hist = klu.macdhist
if klu.macd_state == Chan_MACD_STATE.NEAR0 and len(self.histset_list) > 1:
#print(self.start_time, klu.time, self.near0_count)
self.near0_count += 1
def set_end_klu(self, end_klu, end_type):
self.add_klu(end_klu)
self.end_type = end_type
self.end_klu = end_klu
self.end_time = end_klu.time
K线动能理论.txt
+2 -2
View File
@@ -15,9 +15,9 @@ MACD黄白线和零轴的几种形态：
离开零轴
当MACD黄白线穿过零轴那么进入第一阶段离开零轴，此时能量柱变化越来越大，不断增长，k线加速上涨
高位
当MACD黄白线离开零轴，到一高点时，能量柱此时处于最大，开始减弱时MACD处于高位
当MACD黄白线离开零轴，到一高点时，能量柱此时处于最大，开始减弱时MACD处于高位，高位时MACD黄白线和能量柱是同向的，高位过后是高位空
高位空
当MACD黄白线处于高位，随着K线出现缓慢上涨或者横盘整理，MACD黄白线保持高位出现平滑横盘走势，此时，MACD的能量柱出现衰减变化，同时能量柱喝黄白线之间形成一定的空间夹脚，随着能量柱的不断衰减就导致黄白线喝能量柱之间的空间夹脚越来越大，因此就形成高位空
当MACD黄白线处于高位，随着K线出现缓慢上涨或者横盘整理，MACD黄白线保持高位出现平滑横盘走势，此时，MACD的能量柱出现衰减变化，同时能量柱和黄白线之间形成一定的空间夹脚，随着能量柱的不断衰减就导致黄白线和能量柱之间的空间夹脚越来越大，因此就形成高位空
归零轴
当MACD黄白线在高位，驱动K线上涨的能量所产生的加速度小于或者等于零，K线减速上涨或者下跌，能量变化越来越小，能量柱呈现出一根比一根短的排列方式
穿零轴
config/ChanLun_MACD.json
+83
View File
@@ -0,0 +1,83 @@
{
"$schema": "https://schema.freqtrade.io/schema.json",
"max_open_trades": 1,
"stake_currency": "USDT",
"stake_amount": "unlimited",
"tradable_balance_ratio": 0.99,
"fiat_display_currency": "USD",
"dry_run": true,
"db_url": "sqlite:///tradesv3.chanlun_btc_15.sqlite",
"dry_run_wallet": 1000,
"cancel_open_orders_on_exit": true,
"trading_mode": "futures",
"margin_mode": "isolated",
"can_short" : true,
"timeframe" : "5m",
"process_only_new_candles" : false,
"unfilledtimeout": {
"entry": 1,
"exit": 1,
"exit_timeout_count": 5,
"unit": "minutes"
},
"entry_pricing": {
"price_side": "other",
"use_order_book": false,
"order_book_top": 1,
"price_last_balance": 0.0,
"check_depth_of_market": {
"enabled": false,
"bids_to_ask_delta": 1
}
},
"exit_pricing":{
"price_side": "other",
"use_order_book": false,
"order_book_top": 1
},
"exchange": {
"name": "binance",
"key": "hvoXanRExQvcN4tyGFvEnsSF4gqxXp6ZJnBu5lnhvlVuHaDbj2PhLBQGCLkkyeI8",
"secret": "3UKA2oyDj7OoXrausmnaLwLlNfXmlNf2imBdmQqqKHArcJfk6X9xjaUF19wzu82l",
"ccxt_config": {},
"ccxt_async_config": {},
"pair_whitelist": [
"BTC/USDT:USDT"
],
"pair_blacklist": [
"BNB/.*"
]
},
"pairlists": [
{
"method": "StaticPairList",
"number_assets": 1,
"sort_key": "quoteVolume",
"min_value": 0,
"refresh_period": 1800
}
],
"telegram": {
"enabled": true,
"token": "7677670958:AAFL_jgZvNUTPR3R3vWieREX_tDVi9w2C1Y",
"chat_id": "580807463"
},
"api_server": {
"enabled": true,
"listen_ip_address": "127.0.0.1",
"listen_port": 8813,
"verbosity": "error",
"enable_openapi": false,
"jwt_secret_key": "14d3510740e2c39a973a8895f1aa2704d98d08b86170260085709fa5ea48251d",
"ws_token": "dtKKDnafBrX4icq_ZCw7acJTahTK4h_yvg",
"CORS_origins": [],
"username": "freqtrader",
"password": "FreqTrade007"
},
"bot_name": "freqtrade",
"initial_state": "running",
"force_entry_enable": false,
"internals": {
"process_throttle_secs": 2
}
}
strategies/ChanLun_MACD.py
+346
View File
@@ -0,0 +1,346 @@
# --- Do not remove these libs ---
from statistics import median
from freqtrade.strategy import IStrategy, stoploss_from_absolute
import sys
import os
# 添加父目录到系统路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ChanLun import ChanLun
from ChanLun_Classifier import ChanLunClassifier
from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX
from ChanPY import ChanPY
# --------------------------------
from technical.util import resample_to_interval, resampled_merge
import talib.abstract as ta
import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime, timedelta
from freqtrade.persistence import Trade, Order
from typing import Optional
import logging
logger = logging.getLogger(__name__)
### Now you can use logger.info('asfd') to log
# freqtrade plot-dataframe --strategy ChanLun_MACD --datadir user_data/data/binance -c ./user_data/ChanLun_MACD.json --timerange=20250309-
# freqtrade trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250812-
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_MACD.json -t 5m --pairs BTC/USDT:USDT --timerange=20240101-
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_MACD.json -e 200 --timerange=20250201-20250401
# sudo docker compose run --rm chanlun_btc backtesting -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies --timerange=20250721-
# sudo docker compose run --rm chanlun_btc download-data -c ./user_data/Chan/config/ChanLun_MACD.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101-
# sudo docker compose run --rm chanlun_btc trade -c ./user_data/Chan/config/ChanLun_MACD.json --strategy ChanLun_MACD --strategy-path ./user_data/Chan/strategies
class ChanLun_MACD(IStrategy):
# 标准 Freqtrade 策略：使用 归零轴 + 背离/隐性形态 进行交易
INTERFACE_VERSION: int = 3
# 基本参数
timeframe = '5m'
startup_candle_count = 300
can_short = True
# ROI/止损（止损由自定义 ATR 控制，此处设大）
minimal_roi = {"0": 0.1}
stoploss = -0.3
use_custom_stoploss = True
# 使用市价单，避免回测限价成交不充分导致信号丢单
order_types = {
"entry": "market",
"exit": "market",
"stoploss": "market",
"stoploss_on_exchange": False,
"stoploss_on_exchange_interval": 60,
}
# 过滤：ATR 太小不进场
# 为确保先跑出单，暂不限制 ATR（回测确认后再收紧）
min_atr_value = 0.0
# 可调参数
eps_zero_param = 0.06
div_shift = 2
zero_recent_lookback = 3
# ============ 指标计算 ============
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# MACD
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
# EMA24/EMA52（若无ema24，使用ema26近似）
dataframe['ema24'] = ta.EMA(dataframe, timeperiod=24)
dataframe['ema52'] = ta.EMA(dataframe, timeperiod=52)
# ATR
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
# 归零轴判定
eps_zero = self.eps_zero_param # 可调
dataframe['zero_cross'] = (dataframe['macd'].shift(1) * dataframe['macd'] <= 0)
dataframe['zero_near'] = (dataframe['macd'].abs() <= eps_zero) | ((dataframe['macd'].abs() <= eps_zero) & (dataframe['macdsignal'].abs() <= eps_zero))
dataframe['zero_axis'] = dataframe['zero_cross'] | dataframe['zero_near']
# 价格触碰均线
band24 = 0.006 # 放宽贴近阈值
band52 = 0.008
dataframe['near_ema24'] = (dataframe['ema24'] > 0) & ((dataframe['close'] - dataframe['ema24']).abs() / dataframe['ema24'] <= band24)
dataframe['near_ema52'] = (dataframe['ema52'] > 0) & ((dataframe['close'] - dataframe['ema52']).abs() / dataframe['ema52'] <= band52)
# 背离/隐性背离（可调间隔），先简化到 MACD 快线
sh = self.div_shift
dataframe['bull_div'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0)
dataframe['bear_div'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0)
dataframe['hidden_bull'] = (dataframe['close'] > dataframe['close'].shift(sh)) & (dataframe['macd'] < dataframe['macd'].shift(sh)) & (dataframe['macd'] < 0)
dataframe['hidden_bear'] = (dataframe['close'] < dataframe['close'].shift(sh)) & (dataframe['macd'] > dataframe['macd'].shift(sh)) & (dataframe['macd'] > 0)
# 近N根出现过归零轴（解决“同一根同时满足”过严问题）
zr = dataframe['zero_axis']
for i in range(1, self.zero_recent_lookback):
zr = zr | dataframe['zero_axis'].shift(i)
dataframe['zero_recent'] = zr.fillna(False)
# 近零处快线穿越慢线（补充触发源）
near_zero_now = dataframe['macd'].abs() <= (eps_zero * 2)
cross_up = (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1))
cross_dn = (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1))
dataframe['zero_cross_up_near'] = near_zero_now & cross_up
dataframe['zero_cross_dn_near'] = near_zero_now & cross_dn
# ====== 结构：基于 ChanMACD 的 UnitTF 结束点（与零轴定义一致） ======
try:
from ChanKLU import ChanKLU
from ChanMACD import ChanMACD
klu_list = []
prev = None
for idx, row in dataframe.iterrows():
klu = ChanKLU(
time=idx,
open=float(row.get('open', 0) or 0),
high=float(row.get('high', 0) or 0),
low=float(row.get('low', 0) or 0),
close=float(row.get('close', 0) or 0),
volume=float(row.get('volume', 0) or 0),
)
klu.set_pre(prev)
if prev:
prev.set_next(klu)
klu.ema24 = float(row.get('ema24', 0) or 0)
klu.ema52 = float(row.get('ema52', 0) or 0)
klu.set_indicators({'macd': row.get('macd'), 'macdsignal': row.get('macdsignal'), 'macdhist': row.get('macdhist')})
klu.set_idx(len(klu_list))
klu_list.append(klu)
prev = klu
cm = ChanMACD(klu_list)
end_up = {}
end_dn = {}
for u in cm.unittf_list:
if not getattr(u, 'end_klu', None):
continue
dirv = getattr(u, 'dir', 0)
timev = u.end_klu.time
if dirv >= 0:
end_up[timev] = True
else:
end_dn[timev] = True
dataframe['unit_end_up'] = dataframe.index.to_series().apply(lambda t: bool(end_up.get(t, False))).astype(bool)
dataframe['unit_end_dn'] = dataframe.index.to_series().apply(lambda t: bool(end_dn.get(t, False))).astype(bool)
except Exception:
dataframe['unit_end_up'] = False
dataframe['unit_end_dn'] = False
# ====== 高位空 / 低位多 形态（高位横盘柱衰/低位横盘柱回升） ======
eps_high = max(eps_zero * 2, 0.08)
H = 5
N = 3
macd_high = (dataframe['macd'] > eps_high)
macd_low = (dataframe['macd'] < -eps_high)
# 黄白线高/低位区（结合快慢线）
dataframe['macd_high_zone'] = (dataframe['macd'] > eps_high) & (dataframe['macdsignal'] > eps_high)
dataframe['macd_low_zone'] = (dataframe['macd'] < -eps_high) & (dataframe['macdsignal'] < -eps_high)
hist_down = (dataframe['macdhist'].diff() < 0)
hist_up = (dataframe['macdhist'].diff() > 0)
dataframe['hs_window'] = macd_high.rolling(H).sum() == H
dataframe['ls_window'] = macd_low.rolling(H).sum() == H
dataframe['hist_down_streak'] = hist_down.rolling(N).sum() == N
dataframe['hist_up_streak'] = hist_up.rolling(N).sum() == N
dataframe['high_short_setup'] = (dataframe['hs_window'] & dataframe['hist_down_streak']).fillna(False)
dataframe['low_long_setup'] = (dataframe['ls_window'] & dataframe['hist_up_streak']).fillna(False)
# ====== 基于枢轴点(局部高低点)的直方图背离/隐性背离检测 ======
# 枢轴点定义：高点 high[i] > high[i-1] 且 >= high[i+1]；低点相反
pivot_high = (dataframe['high'] > dataframe['high'].shift(1)) & (dataframe['high'] >= dataframe['high'].shift(-1))
pivot_low = (dataframe['low'] < dataframe['low'].shift(1)) & (dataframe['low'] <= dataframe['low'].shift(-1))
# 直方图峰/谷
hist_peak = (dataframe['macdhist'] > dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] >= dataframe['macdhist'].shift(-1))
hist_trough = (dataframe['macdhist'] < dataframe['macdhist'].shift(1)) & (dataframe['macdhist'] <= dataframe['macdhist'].shift(-1))
# 仅在对应象限判定
hist_peak_pos = hist_peak & (dataframe['macd'] > 0)
hist_trough_neg = hist_trough & (dataframe['macd'] < 0)
# 抽取序列上的上一枢轴值
ph_price = dataframe['high'].where(pivot_high)
pl_price = dataframe['low'].where(pivot_low)
ph_hist = dataframe['macdhist'].where(hist_peak_pos)
pl_hist = dataframe['macdhist'].where(hist_trough_neg)
prev_ph_price = ph_price.shift(1).ffill()
prev_pl_price = pl_price.shift(1).ffill()
prev_ph_hist = ph_hist.shift(1).ffill()
prev_pl_hist = pl_hist.shift(1).ffill()
# 经典背离
bear_div_pivot = pivot_high & hist_peak_pos & (dataframe['high'] > prev_ph_price) & (dataframe['macdhist'] < prev_ph_hist)
bull_div_pivot = pivot_low & hist_trough_neg & (dataframe['low'] < prev_pl_price) & (dataframe['macdhist'] > prev_pl_hist)
# 隐性背离（顺势）
hidden_bear_pivot = pivot_high & hist_peak_pos & (dataframe['high'] < prev_ph_price) & (dataframe['macdhist'] > prev_ph_hist)
hidden_bull_pivot = pivot_low & hist_trough_neg & (dataframe['low'] > prev_pl_price) & (dataframe['macdhist'] < prev_pl_hist)
dataframe['bear_div_pivot'] = bear_div_pivot.fillna(False)
dataframe['bull_div_pivot'] = bull_div_pivot.fillna(False)
dataframe['hidden_bear_pivot'] = hidden_bear_pivot.fillna(False)
dataframe['hidden_bull_pivot'] = hidden_bull_pivot.fillna(False)
# ====== 统计日志（便于回测定位信号规模） ======
try:
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
cnt_zero_recent = int(dataframe['zero_recent'].fillna(False).sum())
cnt_zcup = int(dataframe['zero_cross_up_near'].fillna(False).sum())
cnt_zcdn = int(dataframe['zero_cross_dn_near'].fillna(False).sum())
cnt_bull_div = int(dataframe['bull_div'].fillna(False).sum())
cnt_bear_div = int(dataframe['bear_div'].fillna(False).sum())
cnt_hbull = int(dataframe['hidden_bull'].fillna(False).sum())
cnt_hbear = int(dataframe['hidden_bear'].fillna(False).sum())
cnt_u_end_up = int(dataframe.get('unit_end_up', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_u_end_dn = int(dataframe.get('unit_end_dn', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_hs = int(dataframe.get('high_short_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_ll = int(dataframe.get('low_long_setup', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_bear_div_p = int(dataframe.get('bear_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_bull_div_p = int(dataframe.get('bull_div_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_hbear_p = int(dataframe.get('hidden_bear_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
cnt_hbull_p = int(dataframe.get('hidden_bull_pivot', pd.Series(False, index=dataframe.index)).fillna(False).sum())
zones_high = int(dataframe.get('macd_high_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum())
zones_low = int(dataframe.get('macd_low_zone', pd.Series(False, index=dataframe.index)).fillna(False).sum())
logger.info(f"[{pair}] IND zr={cnt_zero_recent} zcup={cnt_zcup} zcdn={cnt_zcdn} div(bull={cnt_bull_div},bear={cnt_bear_div},hb={cnt_hbull},hs={cnt_hbear}) piv(bull={cnt_bull_div_p},bear={cnt_bear_div_p},hb={cnt_hbull_p},hs={cnt_hbear_p}) zones(high={zones_high},low={zones_low}) unit_end(up={cnt_u_end_up},dn={cnt_u_end_dn}) setup(hs={cnt_hs},ll={cnt_ll})")
except Exception:
pass
# 进场模板（在 populate_entry_trend 中使用）
return dataframe
# ============ 入场/出场信号 ============
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe['enter_long'] = 0
dataframe['enter_short'] = 0
# 做多：柱形枢轴底背离/隐性多 + 低位区 或 近零轴（允许只要枢轴+近零即可）
s_false = pd.Series(False, index=dataframe.index)
bull_hist = (
dataframe.get('bull_div_pivot', s_false).fillna(False)
|
dataframe.get('hidden_bull_pivot', s_false).fillna(False)
)
low_zone = dataframe.get('macd_low_zone', s_false).fillna(False)
# 放宽近零阈值以确保产生成交
near_zero = (dataframe['macd'].abs() <= (self.eps_zero_param * 2.0)).fillna(False)
long_cond = (bull_hist & (low_zone | near_zero | dataframe['zero_recent']))
dataframe.loc[long_cond, 'enter_long'] = 1
# 做空：柱形枢轴顶背离/隐性空 + 高位区 或 近零轴（允许只要枢轴+近零即可）
bear_hist = (
dataframe.get('bear_div_pivot', s_false).fillna(False)
|
dataframe.get('hidden_bear_pivot', s_false).fillna(False)
)
high_zone = dataframe.get('macd_high_zone', s_false).fillna(False)
short_cond = (bear_hist & (high_zone | near_zero | dataframe['zero_recent']))
dataframe.loc[short_cond, 'enter_short'] = 1
# ====== 入场标签与统计（聚焦柱形+区位） ======
try:
long_highzone = (bull_hist & low_zone).fillna(False)
long_nearzero = (bull_hist & near_zero).fillna(False)
short_highzone = (bear_hist & high_zone).fillna(False)
short_nearzero = (bear_hist & near_zero).fillna(False)
dataframe['enter_tag'] = ''
dataframe['long_tag_tmp'] = np.select(
[long_highzone, long_nearzero],
['HIST_BULL_DIV_HIGHZONE', 'HIST_BULL_DIV_NEARZERO'],
default=''
)
dataframe['short_tag_tmp'] = np.select(
[short_highzone, short_nearzero],
['HIST_BEAR_DIV_HIGHZONE', 'HIST_BEAR_DIV_NEARZERO'],
default=''
)
dataframe.loc[dataframe['enter_long'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_long'] == 1, 'long_tag_tmp'].replace('', 'OTHER')
dataframe.loc[dataframe['enter_short'] == 1, 'enter_tag'] = dataframe.loc[dataframe['enter_short'] == 1, 'short_tag_tmp'].replace('', 'OTHER')
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
cnt_long = int((dataframe['enter_long'] == 1).sum())
cnt_short = int((dataframe['enter_short'] == 1).sum())
cnt_l_hz = int(long_highzone.sum()); cnt_l_nz = int(long_nearzero.sum())
cnt_s_hz = int(short_highzone.sum()); cnt_s_nz = int(short_nearzero.sum())
# 最终可下单信号数量（enter_* 列）
el = int((dataframe.get('enter_long', 0) == 1).sum())
es = int((dataframe.get('enter_short', 0) == 1).sum())
logger.info(f"[{pair}] SIG long={cnt_long} short={cnt_short} long_parts(hz={cnt_l_hz},nz={cnt_l_nz}) short_parts(hz={cnt_s_hz},nz={cnt_s_nz}) ENTER(el={el},es={es})")
except Exception:
pass
# 不追加 UnitTF 入场，聚焦柱形+区位组合
try:
idx_long = list(dataframe.index[dataframe['enter_long'] == 1])
idx_short = list(dataframe.index[dataframe['enter_short'] == 1])
def _fmt(ts_list):
return [str(ts_list[i]) for i in range(min(5, len(ts_list)))] + (["..."] if len(ts_list) > 10 else []) + [str(ts_list[i]) for i in range(max(0, len(ts_list)-5), len(ts_list))] if ts_list else []
pair = metadata.get('pair', 'N/A') if isinstance(metadata, dict) else 'N/A'
logger.info(f"[{pair}] ENTER_LONG idx samples: {_fmt(idx_long)}")
logger.info(f"[{pair}] ENTER_SHORT idx samples: {_fmt(idx_short)}")
except Exception:
pass
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
dataframe['exit_long'] = 0
dataframe['exit_short'] = 0
# 退出：MACD 反向穿越零轴 或 触碰 EMA52 失败
long_exit = (dataframe['macd'] < 0) | (dataframe['near_ema52'] & (dataframe['macd'] < dataframe['macdsignal']))
short_exit = (dataframe['macd'] > 0) | (dataframe['near_ema52'] & (dataframe['macd'] > dataframe['macdsignal']))
dataframe.loc[long_exit, 'exit_long'] = 1
dataframe.loc[short_exit, 'exit_short'] = 1
return dataframe
# ============ 过滤与止损 ============
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str | None,
side: str, **kwargs) -> bool:
# 暂时放开所有过滤，确保先产生成交，再逐步收紧
return True
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
# 首次进场保存 ATR 作为 1x 止损距离
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
if dataframe is None or len(dataframe) == 0:
return None
last = dataframe.iloc[-1].squeeze()
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
entry_atr = float(last.get('atr', 0) or 0)
trade.set_custom_data(key="entry_atr", value=entry_atr)
logger.info(f"保存开仓ATR: {entry_atr}")
return None
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, after_fill: bool,
**kwargs) -> float | None:
# 1x ATR 止损
entry_atr = trade.get_custom_data(key="entry_atr")
if entry_atr is None:
# 兜底：5%
return -0.05
if trade.is_short:
stop_price = trade.open_rate + float(entry_atr)
else:
stop_price = trade.open_rate - float(entry_atr)
return stoploss_from_absolute(stop_price, current_rate, is_short=trade.is_short)
web/app.py
+267 -3
View File
@@ -18,8 +18,9 @@ import numpy as np
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from ChanLun import ChanLun
from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE
from ChanEnum import Chan_BI_DIR, Chan_SEG_DIR, Chan_KLC_FX, Chan_FX_TYPE, Chan_MACDSEG_DIR, Chan_MACDHISTSET_DIR
from cn_stock_data import ChinaStockData
from ChanMACD import ChanMACD
# 添加买卖点枚举类型
class TRADE_POINT_TYPE:
@@ -363,6 +364,58 @@ def analyze_chan(df):
bi.cal_macd_div()
#print(bi.start_time, bi.macd_hist, bi.macd_div)
# 添加ChanMACD分析
chan_macd = None
chan_macd_data = {}
try:
# 直接使用ChanLun的get_klu_list方法获取KLU列表
klu_list = chan.get_klu_list(df)
if klu_list and len(klu_list) > 0:
print(f"获取到KLU列表，长度: {len(klu_list)}")
chan_macd = ChanMACD(klu_list)
chan_macd_data = {
'seg_list': chan_macd.seg_list,
'unittf_list': chan_macd.unittf_list,
'histset_list': chan_macd.histset_list,
'high_position_list': chan_macd.high_position_list,
'high_empty_list': chan_macd.high_empty_list,
'low_position_list': getattr(chan_macd, 'low_position_list', []),
'low_empty_list': getattr(chan_macd, 'low_empty_list', []),
'return_zero_list': chan_macd.return_zero_list,
'cross0_up_list': chan_macd.cross0_up_list,
'cross0_down_list': chan_macd.cross0_down_list
}
print(f"ChanMACD分析完成: seg={len(chan_macd.seg_list)}, unittf={len(chan_macd.unittf_list)}, histset={len(chan_macd.histset_list)}")
else:
print("未能获取KLU列表或列表为空")
chan_macd_data = {
'seg_list': [],
'unittf_list': [],
'histset_list': [],
'high_position_list': [],
'high_empty_list': [],
'return_zero_list': [],
'cross0_up_list': [],
'cross0_down_list': []
}
except Exception as e:
print(f"ChanMACD分析出错: {e}")
import traceback
traceback.print_exc()
chan_macd_data = {
'seg_list': [],
'unittf_list': [],
'histset_list': [],
'high_position_list': [],
'high_empty_list': [],
'low_position_list': [],
'low_empty_list': [],
'return_zero_list': [],
'cross0_up_list': [],
'cross0_down_list': []
}
# 获取原始K线数据用于KLU分型分析
klu_list = []
try:
@@ -490,7 +543,8 @@ def analyze_chan(df):
'zs_list': zs_list,
'trade_points': buy_sell_points,
'klc_fx_info': klc_fx_info, # KLC分型信息
'klu_fx_info': klu_fx_info # 添加KLU分型信息
'klu_fx_info': klu_fx_info, # 添加KLU分型信息
'chan_macd': chan_macd_data # 添加ChanMACD分析数据
}
def generate_replay_data(df, client_tz, symbol=None, element_timeframe=None, start_time=None, end_time=None):
@@ -981,6 +1035,211 @@ def format_time_safely(time_obj, client_tz):
# 已经是datetime对象
return time_obj.astimezone(client_tz).isoformat()
def serialize_chan_macd_data(chan_macd_data, client_tz):
"""序列化ChanMACD数据为JSON可序列化格式"""
serialized_data = {
'seg_list': [],
'unittf_list': [],
'histset_list': [],
# 状态标记数据
'high_position_list': [],
'high_empty_list': [],
'low_position_list': [],
'low_empty_list': [],
'return_zero_list': [],
'cross0_up_list': [],
'cross0_down_list': []
}
# 序列化seg_list
for seg in chan_macd_data.get('seg_list', []):
try:
seg_data = {
'start_time': format_time_safely(seg.start_time, client_tz),
'end_time': format_time_safely(seg.end_time, client_tz) if seg.end_time else None,
'seg_dir': 'ABOVE' if seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else 'UNDER',
'klu_count': len(seg.klu_list) if hasattr(seg, 'klu_list') else 0,
'unittf_count': len(seg.unittf_list) if hasattr(seg, 'unittf_list') else 0,
'histset_count': len(seg.hist_set) if hasattr(seg, 'hist_set') else 0
}
serialized_data['seg_list'].append(seg_data)
except Exception as e:
print(f"序列化seg出错: {e}")
continue
# 序列化unittf_list（兼容新结构与枚举类型）
for unittf in chan_macd_data.get('unittf_list', []):
try:
dir_value = getattr(unittf, 'uinttf_dir', None)
dir_name = getattr(dir_value, 'name', dir_value if isinstance(dir_value, str) else None)
start_t = getattr(unittf, 'start_type', None)
start_type = getattr(start_t, 'name', start_t)
end_t = getattr(unittf, 'end_type', None)
end_type = getattr(end_t, 'name', end_t)
peak_abs = getattr(unittf, 'peak_abs', None)
if peak_abs is None:
peak_abs = getattr(unittf, 'peak_hist', None)
length = getattr(unittf, 'length', None)
if length is None:
length = len(unittf.klu_list) if hasattr(unittf, 'klu_list') else None
unittf_data = {
'start_time': format_time_safely(getattr(unittf, 'start_time', None), client_tz),
'end_time': format_time_safely(getattr(unittf, 'end_time', None), client_tz) if getattr(unittf, 'end_time', None) else None,
'dir': dir_name, # 'ABOVE' | 'UNDER' | None
'start_type': start_type, # e.g. 'START' | 'CROSS0' | 'NEAR0_UP' | 'NEAR0_DOWN'
'end_type': end_type,
'invalid': getattr(unittf, 'invalid', False),
'peak_abs': peak_abs,
'length': length,
'klu_count': len(unittf.klu_list) if hasattr(unittf, 'klu_list') else 0,
'histset_count': len(unittf.histset_list) if hasattr(unittf, 'histset_list') else 0
}
serialized_data['unittf_list'].append(unittf_data)
except Exception as e:
print(f"序列化unittf出错: {e}")
continue
# 序列化histset_list
for histset in chan_macd_data.get('histset_list', []):
try:
histset_data = {
'start_time': format_time_safely(getattr(histset, 'start_time', None), client_tz),
'end_time': format_time_safely(getattr(histset, 'end_time', None), client_tz),
'histset_dir': 'ABOVE' if histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE else 'UNDER',
'klu_count': len(histset.klu_list) if hasattr(histset, 'klu_list') else 0
}
serialized_data['histset_list'].append(histset_data)
except Exception as e:
print(f"序列化histset出错: {e}")
continue
# 序列化状态标记数据
# 序列化高位列表
for high_pos in chan_macd_data.get('high_position_list', []):
try:
high_pos_data = {
'time': format_time_safely(high_pos['time'], client_tz),
'end_time': format_time_safely(high_pos.get('end_time'), client_tz) if high_pos.get('end_time') else None,
'type': high_pos.get('type', 'start'),
'macd': high_pos.get('macd'),
'signal': high_pos.get('signal'),
'macdhist': high_pos.get('macdhist'),
'end_macd': high_pos.get('end_macd'),
'end_signal': high_pos.get('end_signal'),
'end_macdhist': high_pos.get('end_macdhist')
}
serialized_data['high_position_list'].append(high_pos_data)
except Exception as e:
print(f"序列化high_position出错: {e}")
continue
# 序列化高位空列表
for high_empty in chan_macd_data.get('high_empty_list', []):
try:
high_empty_data = {
'time': format_time_safely(high_empty['time'], client_tz),
'end_time': format_time_safely(high_empty.get('end_time'), client_tz) if high_empty.get('end_time') else None,
'type': high_empty.get('type', 'start'),
'macd': high_empty.get('macd'),
'signal': high_empty.get('signal'),
'macdhist': high_empty.get('macdhist'),
'end_macd': high_empty.get('end_macd'),
'end_signal': high_empty.get('end_signal'),
'end_macdhist': high_empty.get('end_macdhist')
}
serialized_data['high_empty_list'].append(high_empty_data)
except Exception as e:
print(f"序列化high_empty出错: {e}")
continue
# 序列化低位与低位空
for low_pos in chan_macd_data.get('low_position_list', []):
try:
low_pos_data = {
'time': format_time_safely(low_pos['time'], client_tz),
'end_time': format_time_safely(low_pos.get('end_time'), client_tz) if low_pos.get('end_time') else None,
'type': low_pos.get('type', 'start'),
'macd': low_pos.get('macd'),
'signal': low_pos.get('signal'),
'macdhist': low_pos.get('macdhist'),
'end_macd': low_pos.get('end_macd'),
'end_signal': low_pos.get('end_signal'),
'end_macdhist': low_pos.get('end_macdhist')
}
serialized_data['low_position_list'].append(low_pos_data)
except Exception as e:
print(f"序列化low_position出错: {e}")
continue
for low_empty in chan_macd_data.get('low_empty_list', []):
try:
low_empty_data = {
'time': format_time_safely(low_empty['time'], client_tz),
'end_time': format_time_safely(low_empty.get('end_time'), client_tz) if low_empty.get('end_time') else None,
'type': low_empty.get('type', 'start'),
'macd': low_empty.get('macd'),
'signal': low_empty.get('signal'),
'macdhist': low_empty.get('macdhist'),
'end_macd': low_empty.get('end_macd'),
'end_signal': low_empty.get('end_signal'),
'end_macdhist': low_empty.get('end_macdhist')
}
serialized_data['low_empty_list'].append(low_empty_data)
except Exception as e:
print(f"序列化low_empty出错: {e}")
continue
# 序列化归零轴列表
for return_zero in chan_macd_data.get('return_zero_list', []):
try:
return_zero_data = {
'time': format_time_safely(return_zero['time'], client_tz),
'end_time': format_time_safely(return_zero.get('end_time'), client_tz) if return_zero.get('end_time') else None,
'type': return_zero.get('type', 'start'),
'macd': return_zero.get('macd'),
'signal': return_zero.get('signal'),
'macdhist': return_zero.get('macdhist'),
'end_macd': return_zero.get('end_macd'),
'end_signal': return_zero.get('end_signal'),
'end_macdhist': return_zero.get('end_macdhist')
}
serialized_data['return_zero_list'].append(return_zero_data)
except Exception as e:
print(f"序列化return_zero出错: {e}")
continue
# 序列化穿越零轴列表
for cross0_up in chan_macd_data.get('cross0_up_list', []):
try:
cross0_up_data = {
'time': format_time_safely(cross0_up['time'], client_tz),
'type': cross0_up.get('type', 'start'),
'macd': cross0_up.get('macd'),
'signal': cross0_up.get('signal'),
'macdhist': cross0_up.get('macdhist')
}
serialized_data['cross0_up_list'].append(cross0_up_data)
except Exception as e:
print(f"序列化cross0_up出错: {e}")
continue
for cross0_down in chan_macd_data.get('cross0_down_list', []):
try:
cross0_down_data = {
'time': format_time_safely(cross0_down['time'], client_tz),
'type': cross0_down.get('type', 'start'),
'macd': cross0_down.get('macd'),
'signal': cross0_down.get('signal'),
'macdhist': cross0_down.get('macdhist')
}
serialized_data['cross0_down_list'].append(cross0_down_data)
except Exception as e:
print(f"序列化cross0_down出错: {e}")
continue
return serialized_data
def is_smaller_timeframe(tf1, tf2):
"""判断时间周期tf1是否小于tf2"""
# 定义时间周期的分钟数映射
@@ -1439,7 +1698,9 @@ def analyze():
'fx_strength_level': str(point['fx_strength_level']), # 分型强度等级
'is_strong_fx': bool(point['is_strong_fx']), # 是否为强分型
'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认
} for point in analysis_result['klu_fx_info']]
} for point in analysis_result['klu_fx_info']],
# 添加ChanMACD分析数据
'chan_macd': serialize_chan_macd_data(analysis_result.get('chan_macd', {}), client_tz)
})
# 如果生成了回放数据，添加到返回结果中
@@ -1563,6 +1824,9 @@ def analyze():
'fx_confirmed': bool(point['fx_confirmed']) # 分型是否确认
} for point in element_analysis['klu_fx_info']]
# 添加次周期ChanMACD分析数据
result['element_chan_macd'] = serialize_chan_macd_data(element_analysis.get('chan_macd', {}), client_tz)
pass
return jsonify(result)
web/templates/index.html
+823 -60
View File
File diff suppressed because it is too large Load Diff