#!/usr/bin/env python3 """一次性迁移：根目录引擎模块 → chanlun/ 包 + 根 shim。""" from __future__ import annotations import re import shutil from pathlib import Path ROOT = Path(__file__).resolve().parents[1] CORE = [ "ChanEnum.py", "ChanCTime.py", "ChanKLU.py", "ChanKLC.py", "ChanBI.py", "ChanSBI.py", "ChanSEG.py", "ChanZS.py", "ChanBIZS.py", "ChanBSP.py", "Chan_FX_Box.py", ] INDICATORS = [ "ChanMACD.py", "ChanMACDHistSet.py", "ChanMACDSeg.py", "ChanMACDUnitTF.py", ] ANALYSIS = [ "ChanZone.py", "ChanLun_Classifier.py", "ChanPivotClassifier.py", "ChanPivotMonitor.py", "ChanHeng.py", "ChanPY.py", "Find_Trend.py", "fx_strength_config.py", ] PIPELINE = { "ChanLun.py": "orchestrator.py", "TF_DF.py": "timeframe.py", } MODULE_PKG: dict[str, str] = {} for name in CORE: MODULE_PKG[name[:-3]] = "chanlun.core" for name in INDICATORS: MODULE_PKG[name[:-3]] = "chanlun.indicators" for name in ANALYSIS: MODULE_PKG[name[:-3]] = "chanlun.analysis" IMPORT_TARGET = { **{k: f"{v}.{k}" for k, v in MODULE_PKG.items()}, "ChanLun": "chanlun.pipeline.orchestrator", "TF_DF": "chanlun.pipeline.timeframe", } KNOWN = set(IMPORT_TARGET) def rewrite_imports(text: str) -> str: """只改写已知缠论模块；绝不拆分 typing/talib 等标准 from-import。""" out_lines = [] for line in text.splitlines(keepends=True): nl = "\n" if line.endswith("\n") else "" raw = line[:-1] if nl else line indent_m = re.match(r"^(\s*)", raw) indent = indent_m.group(1) if indent_m else "" stripped = raw[len(indent) :] if stripped.startswith("from ") and " import " in stripped: m = re.match(r"^from (\S+) import (.*)$", stripped) if m and m.group(1) in KNOWN: raw = f"{indent}from {IMPORT_TARGET[m.group(1)]} import {m.group(2)}" out_lines.append(raw + nl) continue if stripped.startswith("import "): rest = stripped[len("import ") :] # 跳过 import x as y 复合以外的非已知模块整行 parts = [p.strip() for p in rest.split(",")] if not any(p.split(" as ")[0].strip() in KNOWN for p in parts): out_lines.append(line) continue new_parts = [] for p in parts: base = p.split(" as ")[0].strip() if base not in KNOWN: new_parts.append(f"import {p}") continue target = IMPORT_TARGET[base] if " as " in p: new_parts.append( f"import {target} as {p.split(' as ', 1)[1].strip()}" ) elif base in ("ChanLun", "TF_DF"): new_parts.append(f"from {target} import {base}") else: new_parts.append(f"import {target} as {base}") # 多模块拆成多行，保持可读 out_lines.append(nl.join(indent + x for x in new_parts) + nl) continue out_lines.append(line) return "".join(out_lines) def write_shim(mod_name: str): path = ROOT / f"{mod_name}.py" if mod_name == "ChanLun": body = ( '"""兼容 shim — 请优先 from chanlun import ..."""\n' "from chanlun.pipeline.orchestrator import ChanLun # noqa: F401\n" "from chanlun.pipeline.timeframe import TF_DF # noqa: F401\n" ) elif mod_name == "TF_DF": body = ( '"""兼容 shim — 请优先 from chanlun import ..."""\n' "from chanlun.pipeline.timeframe import TF_DF # noqa: F401\n" ) else: target = IMPORT_TARGET[mod_name] body = ( f'"""兼容 shim — 请优先 from chanlun import ..."""\n' f"from {target} import * # noqa: F403\n" ) path.write_text(body, encoding="utf-8") def main(): for sub in ("core", "indicators", "analysis", "pipeline", "pipeline/builders"): d = ROOT / "chanlun" / sub d.mkdir(parents=True, exist_ok=True) (d / "__init__.py").write_text("", encoding="utf-8") moved = [] def move_list(names, dest_pkg: Path): for name in names: src = ROOT / name if not src.exists(): print("skip missing", name) continue dst = dest_pkg / name text = rewrite_imports(src.read_text(encoding="utf-8")) dst.write_text(text, encoding="utf-8") src.unlink() moved.append(name) move_list(CORE, ROOT / "chanlun" / "core") move_list(INDICATORS, ROOT / "chanlun" / "indicators") move_list(ANALYSIS, ROOT / "chanlun" / "analysis") for src_name, dst_name in PIPELINE.items(): src = ROOT / src_name if not src.exists(): continue dst = ROOT / "chanlun" / "pipeline" / dst_name text = rewrite_imports(src.read_text(encoding="utf-8")) dst.write_text(text, encoding="utf-8") src.unlink() moved.append(src_name) # 恢复 get_zs_list（受控 L1） tf_path = ROOT / "chanlun" / "pipeline" / "timeframe.py" tf_text = tf_path.read_text(encoding="utf-8") if "def get_zs_list" not in tf_text: needle = "\tdef calculate_seg_zs(self, seg_list):\n\t\treturn self.get_seg_zs_list(seg_list)\n" insert = ( "\tdef get_zs_list(self, bi_list, seg_list):\n" "\t\t\"\"\"兼容历史 API：线段中枢列表。\"\"\"\n" "\t\treturn self.get_seg_zs_list(seg_list)\n" + needle ) if needle in tf_text: tf_path.write_text(tf_text.replace(needle, insert), encoding="utf-8") else: tf_path.write_text( tf_text + "\n\tdef get_zs_list(self, bi_list, seg_list):\n" + "\t\treturn self.get_seg_zs_list(seg_list)\n", encoding="utf-8", ) print("patched get_zs_list") for name in CORE + INDICATORS + ANALYSIS: write_shim(name[:-3]) write_shim("ChanLun") write_shim("TF_DF") (ROOT / "chanlun" / "__init__.py").write_text( '"""缠论引擎正式包。"""\n' "from chanlun.pipeline.orchestrator import ChanLun\n" "from chanlun.pipeline.timeframe import TF_DF\n" '__all__ = ["ChanLun", "TF_DF"]\n', encoding="utf-8", ) examples = ROOT / "examples" examples.mkdir(exist_ok=True) notes = ROOT / "docs" / "notes" notes.mkdir(parents=True, exist_ok=True) for f in ("fx_strength_example.py", "realtime_fx_example.py"): p = ROOT / f if p.exists(): dest = examples / f if dest.exists(): dest.unlink() shutil.move(str(p), str(dest)) for f in ("缠论.txt", "操作策略.txt", "chanlun.txt", "K线动能理论.txt"): p = ROOT / f if p.exists(): dest = notes / f if dest.exists(): dest.unlink() shutil.move(str(p), str(dest)) tc = ROOT / "test_classifier.py" if tc.exists(): dest = ROOT / "tests" / "test_classifier.py" if dest.exists(): dest.unlink() shutil.move(str(tc), str(dest)) print("moved", len(moved), "modules") if __name__ == "__main__": main()