jackyu66gitandCursor
74dec4e50b
将根目录引擎迁入 chanlun/ 并保留兼容 shim；拆分 TF_DF 与 web 服务； 前端模块化；strategies 改用 chanlun 导入；补充 ESS 文档与 golden 回归。 Co-authored-by: Cursor <cursoragent@cursor.com>
229 lines
7.3 KiB
Python
229 lines
7.3 KiB
Python
#!/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()
|