"""中控「策略说明」:读取 docs/strategy MD + checklists JSON.""" from __future__ import annotations import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any from lib.paths import REPO_ROOT STRATEGY_EXCHANGES: tuple[str, ...] = ("binance", "okx", "gate") STRATEGY_META: dict[str, dict[str, str]] = { "binance": { "label": "币安", "title": "币安·山寨多头趋势", "md_file": "binance-alt-trend-long.md", }, "okx": { "label": "OKX", "title": "OKX·多空趋势", "md_file": "okx-trend-both.md", }, "gate": { "label": "Gate", "title": "Gate·BTC 日内", "md_file": "gate-intraday.md", }, } def _strategy_dir() -> Path: return REPO_ROOT / "docs" / "strategy" def _checklist_path(exchange_key: str) -> Path: return _strategy_dir() / "checklists" / f"{exchange_key.strip().lower()}.json" def _md_path(exchange_key: str) -> Path: meta = STRATEGY_META.get((exchange_key or "").strip().lower()) if not meta: raise KeyError(exchange_key) return _strategy_dir() / meta["md_file"] def _parse_version(md_text: str) -> str: m = re.search(r">\s*\*\*状态\*\*[::]\s*(v[\d.]+)", md_text) if m: return m.group(1) m = re.search(r"\|\s*v([\d.]+)\s*\|", md_text) if m: return f"v{m.group(1)}" return "" def render_markdown_html(md_text: str) -> str: try: import markdown # type: ignore return markdown.markdown( md_text, extensions=["tables", "fenced_code", "nl2br", "sane_lists"], ) except Exception: return _simple_md_html(md_text) def _simple_md_html(md_text: str) -> str: from html import escape lines = md_text.replace("\r\n", "\n").replace("\r", "\n").splitlines() out: list[str] = [] i = 0 in_code = False code_buf: list[str] = [] list_buf: list[str] = [] list_ordered = False def flush_list() -> None: nonlocal list_buf, list_ordered if not list_buf: return tag = "ol" if list_ordered else "ul" out.append(f"<{tag}>") for item in list_buf: out.append(f"
{escape(chr(10).join(code_buf))}")
code_buf = []
in_code = False
while i < len(lines):
line = lines[i]
if line.strip().startswith("```"):
flush_list()
if in_code:
flush_code()
else:
in_code = True
i += 1
continue
if in_code:
code_buf.append(line)
i += 1
continue
if re.match(r"^\s*\|", line) and i + 1 < len(lines) and re.match(r"^\s*\|?\s*[-:| ]+\|", lines[i + 1]):
flush_list()
header = [c.strip() for c in line.strip().strip("|").split("|")]
i += 2
rows: list[list[str]] = []
while i < len(lines) and re.match(r"^\s*\|", lines[i]):
rows.append([c.strip() for c in lines[i].strip().strip("|").split("|")])
i += 1
out.append("| {_inline_md(h)} | " for h in header) + "
|---|
| {_inline_md(c)} | " for c in row) + "
{_inline_md(line.lstrip('>').strip())}") i += 1 continue m = re.match(r"^(\d+)\.\s+(.*)$", line.strip()) if m: if list_buf and not list_ordered: flush_list() list_ordered = True list_buf.append(m.group(2)) i += 1 continue if re.match(r"^[-*]\s+", line.strip()): if list_buf and list_ordered: flush_list() list_ordered = False list_buf.append(re.sub(r"^[-*]\s+", "", line.strip())) i += 1 continue if not line.strip(): flush_list() i += 1 continue flush_list() out.append(f"
{_inline_md(line.strip())}
") i += 1 flush_list() flush_code() return "\n".join(out) def _inline_md(text: str) -> str: from html import escape s = escape(text) s = re.sub(r"`([^`]+)`", r"\1", s)
s = re.sub(r"\*\*([^*]+)\*\*", r"\1", s)
return s
def load_checklist(exchange_key: str) -> dict[str, Any]:
path = _checklist_path(exchange_key)
if not path.is_file():
return {"exchange": exchange_key, "title": "开仓检查清单", "groups": []}
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
return {"exchange": exchange_key, "title": "开仓检查清单", "groups": []}
return data
def load_strategy_payload(exchange_key: str) -> dict[str, Any]:
key = (exchange_key or "").strip().lower()
if key not in STRATEGY_META:
raise KeyError(exchange_key)
meta = STRATEGY_META[key]
md_path = _md_path(key)
md_text = md_path.read_text(encoding="utf-8") if md_path.is_file() else ""
checklist = load_checklist(key)
version = _parse_version(md_text) or str(checklist.get("version") or "")
return {
"ok": True,
"exchange_key": key,
"label": meta["label"],
"title": meta["title"],
"version": version,
"md_source": str(md_path.relative_to(REPO_ROOT)).replace("\\", "/"),
"strategy_html": render_markdown_html(md_text),
"checklist": checklist,
}
def strategy_meta_payload() -> dict[str, Any]:
tabs = [
{"key": k, "label": STRATEGY_META[k]["label"], "title": STRATEGY_META[k]["title"]}
for k in STRATEGY_EXCHANGES
]
return {"ok": True, "exchanges": tabs}
def _checklist_html(checklist: dict[str, Any], *, include_title: bool = True) -> str:
groups = checklist.get("groups") or []
parts: list[str] = []
if include_title:
parts.append(f"