jackyu66gitandCursor
74dec4e50b
将根目录引擎迁入 chanlun/ 并保留兼容 shim；拆分 TF_DF 与 web 服务； 前端模块化；strategies 改用 chanlun 导入；补充 ESS 文档与 golden 回归。 Co-authored-by: Cursor <cursoragent@cursor.com>
222 lines
6.5 KiB
Python
222 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
|
"""将 timeframe.TF_DF 拆为 mixin builders（方法体原样搬移）。"""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import re
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC = ROOT / "chanlun" / "pipeline" / "timeframe.py"
|
|
BUILDERS = ROOT / "chanlun" / "pipeline" / "builders"
|
|
|
|
# method name -> builder module
|
|
ASSIGN = {
|
|
"get_ema52": "indicators",
|
|
"get_ema24": "indicators",
|
|
"add_indicators": "indicators",
|
|
"get_klu_state": "kline",
|
|
"get_ema_state": "indicators",
|
|
"check_fx1": "kline",
|
|
"check_fx": "kline",
|
|
"check_fx2": "kline",
|
|
"check_fx_pattern": "kline",
|
|
"cal_volume_ratio": "kline",
|
|
"cal_kl_data": "kline",
|
|
"get_kl_data": "kline",
|
|
"get_klc_list": "kline",
|
|
"get_klu_list": "kline",
|
|
"cal_klu_pattern": "kline",
|
|
"_detect_single_reversal_pattern": "kline",
|
|
"_detect_double_pattern": "kline",
|
|
"_detect_triple_pattern": "kline",
|
|
"cal_trend": "bi",
|
|
"get_bi_list": "bi",
|
|
"cal_bi_list": "bi",
|
|
"check_top_fx": "bi",
|
|
"check_bottom_fx": "bi",
|
|
"get_bsp_state": "bsp",
|
|
"get_above_zero_bsp": "bsp",
|
|
"find_all_bsp": "bsp",
|
|
"check_bi_div": "bsp",
|
|
"find_first_bsp": "bsp",
|
|
"find_second_bsp": "bsp",
|
|
"get_zs_state": "zs",
|
|
"cal_bi_zs": "zs",
|
|
"cal_bi_zs_list": "zs",
|
|
"get_bi_zs_list": "zs",
|
|
"cal_bi_zs_list_pure": "zs",
|
|
"calculate_seg_zs": "zs",
|
|
"get_seg_zs_list": "zs",
|
|
"get_big_zs_list": "zs",
|
|
"get_zs_list": "zs",
|
|
"get_seg_list": "seg",
|
|
"get_decimal": "indicators",
|
|
}
|
|
|
|
HEADER = '''\
|
|
"""TF_DF builder mixin — 由 split_tfdf_builders 自动生成，逻辑与原 TF_DF 一致。"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import timedelta
|
|
from decimal import Decimal
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import talib.abstract as ta
|
|
from pandas import DataFrame
|
|
from technical.util import resample_to_interval
|
|
|
|
from chanlun.core.ChanBI import ChanBI
|
|
from chanlun.core.ChanBIZS import ChanBIZS
|
|
from chanlun.core.ChanBSP import ChanBSP
|
|
from chanlun.core.ChanEnum import (
|
|
Chan_BI_DIR,
|
|
Chan_BSP_DIR,
|
|
Chan_BSP_TYPE,
|
|
Chan_FX_TYPE,
|
|
Chan_K_DIR,
|
|
Chan_KLC_FX,
|
|
Chan_KLC_STATE,
|
|
Chan_KLINE_DIR,
|
|
Chan_KLU_PATTERN,
|
|
Chan_PRICE_TREND,
|
|
Chan_SEG_DIR,
|
|
Chan_ZS_DIR,
|
|
)
|
|
from chanlun.core.ChanKLC import ChanKLC
|
|
from chanlun.core.ChanKLU import ChanKLU
|
|
from chanlun.core.ChanSBI import ChanSBI
|
|
from chanlun.core.ChanSEG import ChanSEG
|
|
from chanlun.core.ChanZS import ChanZS, ChanZS_Big
|
|
from chanlun.indicators.ChanMACD import ChanMACD
|
|
|
|
|
|
'''
|
|
|
|
|
|
def main():
|
|
source = SRC.read_text(encoding="utf-8")
|
|
# Use tabs; extract method bodies by line scanning
|
|
lines = source.splitlines(keepends=True)
|
|
# find class TF_DF
|
|
class_start = None
|
|
for i, line in enumerate(lines):
|
|
if line.startswith("class TF_DF"):
|
|
class_start = i
|
|
break
|
|
if class_start is None:
|
|
raise SystemExit("TF_DF not found")
|
|
|
|
# collect methods: name -> (start, end exclusive)
|
|
methods = []
|
|
i = class_start + 1
|
|
while i < len(lines):
|
|
line = lines[i]
|
|
m = re.match(r"^\tdef ([A-Za-z_][\w]*)\(", line)
|
|
if m:
|
|
name = m.group(1)
|
|
start = i
|
|
i += 1
|
|
while i < len(lines):
|
|
if re.match(r"^\tdef [A-Za-z_]", lines[i]) or (
|
|
lines[i] and not lines[i].startswith("\t") and not lines[i].startswith(" ") and lines[i].strip()
|
|
):
|
|
break
|
|
# nested def inside method starts with \t\tdef
|
|
i += 1
|
|
methods.append((name, start, i))
|
|
continue
|
|
i += 1
|
|
|
|
by_mod: dict[str, list[str]] = {k: [] for k in ("indicators", "kline", "bi", "seg", "zs", "bsp")}
|
|
keep_in_facade = [] # __init__, init_TF_DF, get_current_klc
|
|
|
|
for name, start, end in methods:
|
|
body = "".join(lines[start:end])
|
|
mod = ASSIGN.get(name)
|
|
if mod is None:
|
|
keep_in_facade.append((name, body))
|
|
else:
|
|
by_mod[mod].append(body)
|
|
|
|
BUILDERS.mkdir(parents=True, exist_ok=True)
|
|
(BUILDERS / "__init__.py").write_text("", encoding="utf-8")
|
|
|
|
mixin_classes = []
|
|
for mod, bodies in by_mod.items():
|
|
class_name = "".join(p.title() for p in mod.split("_")) + "BuilderMixin"
|
|
mixin_classes.append(class_name)
|
|
content = HEADER + f"class {class_name}:\n"
|
|
if not bodies:
|
|
content += "\tpass\n"
|
|
else:
|
|
content += "\n".join(bodies)
|
|
if not content.endswith("\n"):
|
|
content += "\n"
|
|
(BUILDERS / f"{mod}.py").write_text(content, encoding="utf-8")
|
|
print(f"wrote {mod}.py methods={len(bodies)} class={class_name}")
|
|
|
|
# rewrite timeframe.py facade
|
|
facade_imports = '''\
|
|
from datetime import timedelta
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import talib.abstract as ta
|
|
from pandas import DataFrame
|
|
from technical.util import resample_to_interval
|
|
|
|
from chanlun.core.ChanBI import ChanBI
|
|
from chanlun.core.ChanBIZS import ChanBIZS
|
|
from chanlun.core.ChanBSP import ChanBSP
|
|
from chanlun.core.ChanEnum import (
|
|
Chan_BI_DIR,
|
|
Chan_BSP_DIR,
|
|
Chan_BSP_TYPE,
|
|
Chan_FX_TYPE,
|
|
Chan_K_DIR,
|
|
Chan_KLC_FX,
|
|
Chan_KLC_STATE,
|
|
Chan_KLINE_DIR,
|
|
Chan_KLU_PATTERN,
|
|
Chan_PRICE_TREND,
|
|
Chan_SEG_DIR,
|
|
Chan_ZS_DIR,
|
|
)
|
|
from chanlun.core.ChanKLC import ChanKLC
|
|
from chanlun.core.ChanKLU import ChanKLU
|
|
from chanlun.core.ChanSBI import ChanSBI
|
|
from chanlun.core.ChanSEG import ChanSEG
|
|
from chanlun.core.ChanZS import ChanZS, ChanZS_Big
|
|
from chanlun.indicators.ChanMACD import ChanMACD
|
|
from chanlun.pipeline.builders.bi import BiBuilderMixin
|
|
from chanlun.pipeline.builders.bsp import BspBuilderMixin
|
|
from chanlun.pipeline.builders.indicators import IndicatorsBuilderMixin
|
|
from chanlun.pipeline.builders.kline import KlineBuilderMixin
|
|
from chanlun.pipeline.builders.seg import SegBuilderMixin
|
|
from chanlun.pipeline.builders.zs import ZsBuilderMixin
|
|
|
|
'''
|
|
bases = ", ".join(
|
|
[
|
|
"IndicatorsBuilderMixin",
|
|
"KlineBuilderMixin",
|
|
"BiBuilderMixin",
|
|
"SegBuilderMixin",
|
|
"ZsBuilderMixin",
|
|
"BspBuilderMixin",
|
|
]
|
|
)
|
|
facade = facade_imports + f"class TF_DF({bases}):\n"
|
|
for name, body in keep_in_facade:
|
|
facade += body
|
|
if not body.endswith("\n"):
|
|
facade += "\n"
|
|
SRC.write_text(facade, encoding="utf-8")
|
|
print("rewrote timeframe.py facade, kept", [n for n, _ in keep_in_facade])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|