refactor: 缠论引擎迁入 chan/ 分层解耦,指标外置

将核心结构、指标与分析拆到 chan/{core,indicators,analysis,pipeline};
根目录保留兼容 shim;strategies 改为从 chan 包导入;买卖点经 bsp_macd 与 MACD 接合。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Porter
2026-08-03 14:47:13 +08:00
co-authored by Cursor
parent 2e905e7238
commit 2c1232555e
90 changed files with 9067 additions and 8535 deletions
+113
View File
@@ -0,0 +1,113 @@
"""买卖点 × MACD:几何候选在 core,背驰确认在此接合。"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, List, Optional, Sequence
from ..core.ChanBI import ChanBI
from ..core.ChanBSP import ChanBSP
from ..core.ChanEnum import Chan_BI_DIR, Chan_BSP_DIR, Chan_BSP_TYPE
from ..indicators.store import IndicatorStore
def bi_macd_area(bi: ChanBI, store: Optional[IndicatorStore] = None) -> float:
"""笔内同向 macdhist 累积面积。优先用 IndicatorStore,否则回退 klu.macdhist。"""
area = 0.0
for klc in bi.klc_list:
for klu in klc.klu_list:
if store is not None:
hist = store.get(klu.idx, "macdhist", 0) or 0
else:
hist = getattr(klu, "macdhist", 0) or 0
try:
hist = float(hist)
except (TypeError, ValueError):
hist = 0.0
if bi.dir == Chan_BI_DIR.UP and hist > 0:
area += hist
elif bi.dir == Chan_BI_DIR.DOWN and hist < 0:
area -= hist
return area
def check_bi_div(
zs,
leave_bi: ChanBI,
store: Optional[IndicatorStore] = None,
) -> bool:
"""一类买卖点背驰:离开笔相对进入笔同向 MACD 柱面积收敛。"""
enter_bi = zs.bi_list[0].pre if zs.bi_list else None
if not enter_bi or enter_bi.dir != leave_bi.dir:
return False
leave_area = abs(bi_macd_area(leave_bi, store))
enter_area = abs(bi_macd_area(enter_bi, store))
return leave_area < enter_area
def check_bi_pair_div(
leave_bi: ChanBI,
compare_bi: ChanBI,
store: Optional[IndicatorStore] = None,
) -> bool:
"""两笔同向力度比较(离开笔面积 < 比较笔)。"""
leave_area = abs(bi_macd_area(leave_bi, store))
compare_area = abs(bi_macd_area(compare_bi, store))
return leave_area < compare_area
@dataclass
class ConfirmedBSP:
"""几何买卖点 + MACD 确认结果。"""
bsp: ChanBSP
div_confirmed: bool
hist_area: float = 0.0
compare_hist_area: float = 0.0
score: float = 0.0
macd_state: Any = None
@property
def type(self) -> Chan_BSP_TYPE:
return self.bsp.type
@property
def dir(self) -> Chan_BSP_DIR:
return self.bsp.dir
@property
def bi(self) -> ChanBI:
return self.bsp.bi
def confirm_bsp(
geo_bsp_list: Sequence[ChanBSP],
store: Optional[IndicatorStore] = None,
require_div_for_types: Optional[Sequence[Chan_BSP_TYPE]] = None,
) -> List[ConfirmedBSP]:
"""
将几何 BSP 升格为 ConfirmedBSP。
默认:B1/S1/T1 需要背驰确认;B3/S3 几何即可,div_confirmed=True。
"""
if require_div_for_types is None:
require_div_for_types = (
Chan_BSP_TYPE.B1,
Chan_BSP_TYPE.S1,
)
out: List[ConfirmedBSP] = []
for bsp in geo_bsp_list:
area = bi_macd_area(bsp.bi, store)
need_div = bsp.type in require_div_for_types
if need_div and bsp.zs is not None:
div_ok = check_bi_div(bsp.zs, bsp.bi, store)
else:
div_ok = True
out.append(
ConfirmedBSP(
bsp=bsp,
div_confirmed=div_ok,
hist_area=area,
score=1.0 if div_ok else 0.0,
)
)
return out