crypto_monitor/lib/hub/hub_strategy_lib.py

"""中控「策略说明」:读取 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, ...] = (
    "playbook_v2",
    "playbook",
    "behavior",
    "binance",
    "okx",
    "gate",
)

STRATEGY_META: dict[str, dict[str, str]] = {
    "playbook_v2": {
        "label": "执行手册v2",
        "title": "交易执行手册 v2（期权 / 合约 · 无对冲）",
        "md_rel": "docs/交易执行手册-v2-期权与合约.md",
    },
    "playbook": {
        "label": "执行手册v1",
        "title": "交易执行手册 v1（期权为主 · Gate 为辅 · 含对冲）",
        # 相对仓库根;其余条目用 md_file 相对 docs/strategy
        "md_rel": "docs/交易执行手册-期权与Gate.md",
    },
    "behavior": {
        "label": "行为准则",
        "title": "交易行为准则（开单三检）",
        "md_rel": "docs/交易行为准则-开单三检.md",
    },
    "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)
    md_rel = (meta.get("md_rel") or "").strip()
    if md_rel:
        return REPO_ROOT / md_rel
    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"<li>{_inline_md(item)}</li>")
        out.append(f"</{tag}>")
        list_buf = []

    def flush_code() -> None:
        nonlocal code_buf, in_code
        if not code_buf:
            return
        out.append(f"<pre><code>{escape(chr(10).join(code_buf))}</code></pre>")
        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("<table><thead><tr>" + "".join(f"<th>{_inline_md(h)}</th>" for h in header) + "</tr></thead><tbody>")
            for row in rows:
                out.append("<tr>" + "".join(f"<td>{_inline_md(c)}</td>" for c in row) + "</tr>")
            out.append("</tbody></table>")
            continue
        if re.match(r"^#{1,3}\s+", line):
            flush_list()
            m = re.match(r"^(#{1,3})\s+(.*)$", line)
            if m:
                level = len(m.group(1))
                out.append(f"<h{level}>{_inline_md(m.group(2))}</h{level}>")
            i += 1
            continue
        if line.strip() == "---":
            flush_list()
            out.append("<hr>")
            i += 1
            continue
        if line.startswith(">"):
            flush_list()
            out.append(f"<blockquote>{_inline_md(line.lstrip('>').strip())}</blockquote>")
            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"<p>{_inline_md(line.strip())}</p>")
        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"<code>\1</code>", s)
    s = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", 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,
    }


_STRATEGY_TAB_DISPLAY_PREF: dict[str, str] = {
    "playbook_v2": "show_strategy_playbook_v2",
    "playbook": "show_strategy_playbook",
    "behavior": "show_strategy_behavior",
    "binance": "show_strategy_binance",
    "okx": "show_strategy_okx",
    "gate": "show_strategy_gate",
}


def strategy_meta_payload(display: dict[str, Any] | None = None) -> dict[str, Any]:
    prefs = display if isinstance(display, dict) else {}
    tabs = []
    for k in STRATEGY_EXCHANGES:
        pref_key = _STRATEGY_TAB_DISPLAY_PREF.get(k)
        if pref_key and prefs.get(pref_key) is False:
            continue
        tabs.append(
            {
                "key": k,
                "label": STRATEGY_META[k]["label"],
                "title": STRATEGY_META[k]["title"],
            }
        )
    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"<h2>{escape_html(str(checklist.get('title') or '开仓检查清单'))}</h2>")
    for grp in groups:
        if not isinstance(grp, dict):
            continue
        gtitle = escape_html(str(grp.get("title") or ""))
        parts.append(f"<h3>{gtitle}</h3><ul class=\"checklist\">")
        for item in grp.get("items") or []:
            # 用 CSS 方框代替 ☐,避免部分打印机/预览吞掉 Unicode 勾选符
            parts.append(
                f"<li><span class=\"box\" aria-hidden=\"true\"></span>"
                f"<span class=\"item-text\">{escape_html(str(item))}</span></li>"
            )
        parts.append("</ul>")
    footnotes = checklist.get("footnotes") or []
    if footnotes:
        parts.append("<ul class=\"footnotes\">")
        for note in footnotes:
            parts.append(f"<li>{escape_html(str(note))}</li>")
        parts.append("</ul>")
    return "\n".join(parts)


def _print_document_css() -> str:
    return """
body {
  margin: 0;
  padding: 36px 28px 48px;
  font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC",
    "Hiragino Sans GB", "Microsoft YaHei", sans-serif;
  font-size: 14px;
  line-height: 1.65;
  color: #1a1a1a;
  background: #fff;
}
.doc {
  max-width: 720px;
  margin: 0 auto;
}
.doc-head {
  margin-bottom: 28px;
  padding-bottom: 14px;
  border-bottom: 1px solid #e5e5e5;
}
.doc-head h1 {
  margin: 0 0 8px;
  font-size: 1.55rem;
  font-weight: 600;
  line-height: 1.3;
}
.doc-meta {
  margin: 0;
  color: #666;
  font-size: 0.85rem;
}
.doc-body h2 {
  font-size: 1.12rem;
  margin: 1.6em 0 0.55em;
  font-weight: 600;
}
.doc-body h3 {
  font-size: 1rem;
  margin: 1.2em 0 0.45em;
  font-weight: 600;
}
.doc-body h2:first-child,
.doc-body h3:first-child {
  margin-top: 0;
}
.doc-body p {
  margin: 0.65em 0;
}
.doc-body table {
  border-collapse: collapse;
  width: 100%;
  font-size: 0.92rem;
  margin: 10px 0 14px;
}
.doc-body th,
.doc-body td {
  border: 1px solid #d8d8d8;
  padding: 8px 10px;
  text-align: left;
  vertical-align: top;
}
.doc-body blockquote {
  margin: 12px 0;
  padding: 8px 14px;
  border-left: 4px solid #c8c8c8;
  color: #444;
  background: #fafafa;
}
.doc-body pre,
.doc-body code {
  font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
  font-size: 0.88em;
}
.doc-body pre {
  padding: 10px 12px;
  background: #f6f6f6;
  border: 1px solid #e8e8e8;
  border-radius: 4px;
  overflow-x: auto;
}
.doc-body hr {
  border: none;
  border-top: 1px solid #e5e5e5;
  margin: 1.4em 0;
}
.checklist {
  list-style: none;
  margin: 0 0 16px;
  padding: 0;
}
.checklist li {
  display: flex;
  align-items: flex-start;
  gap: 8px;
  margin: 8px 0;
  padding: 0;
}
.box {
  flex: 0 0 auto;
  display: inline-block;
  width: 0.95em;
  height: 0.95em;
  margin-top: 0.2em;
  border: 1.5px solid #333;
  border-radius: 2px;
  background: #fff;
  box-sizing: border-box;
  -webkit-print-color-adjust: exact;
  print-color-adjust: exact;
}
.item-text {
  flex: 1 1 auto;
  min-width: 0;
}
.footnotes {
  margin: 18px 0 0;
  padding-left: 20px;
  color: #666;
  font-size: 0.85rem;
}
.doc-foot {
  margin-top: 28px;
  padding-top: 12px;
  border-top: 1px dashed #ddd;
  color: #888;
  font-size: 0.78rem;
}
@media print {
  body { padding: 0; }
  .doc { max-width: none; }
  .doc-head { break-after: avoid; }
  .doc-body h2, .doc-body h3 { break-after: avoid; }
  .checklist li { break-inside: avoid; }
}
"""


def _print_auto_script() -> str:
    # 不在 afterprint 里 window.close(): Firefox 会在打开打印框时就触发 afterprint,
    # 短页(检查清单)更容易被立刻关掉,表现为「打不开/打不出来」;策略长文相对不易复现.
    return """<script>
(function () {
  function go() {
    try { window.focus(); } catch (e) {}
    window.setTimeout(function () {
      try { window.print(); } catch (e) {}
    }, 250);
  }
  if (document.readyState === "complete") go();
  else window.addEventListener("load", go, { once: true });
})();
</script>"""


def build_print_html(exchange_key: str, part: str = "doc") -> str:
    """part: doc | checklist"""
    payload = load_strategy_payload(exchange_key)
    key = payload["exchange_key"]
    label = payload["label"]
    title = payload["title"]
    version = payload.get("version") or ""
    checklist = payload.get("checklist") or {}
    now = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M")
    part = (part or "doc").strip().lower()
    css = _print_document_css()

    if part == "checklist":
        cl_title = str(checklist.get("title") or "开仓检查清单")
        page_title = f"{label} · {cl_title}"
        body = f"""<article class="doc">
<header class="doc-head">
  <h1>{escape_html(cl_title)}</h1>
  <p class="doc-meta">{escape_html(label)} · {escape_html(version)} · 打印 {escape_html(now)}</p>
</header>
<div class="doc-body">{_checklist_html(checklist, include_title=False)}</div>
</article>"""
    elif part == "doc":
        page_title = f"{label} · 策略说明"
        source = payload.get("md_source") or ""
        body = f"""<article class="doc">
<header class="doc-head">
  <h1>{escape_html(title)}</h1>
  <p class="doc-meta">{escape_html(label)} · {escape_html(version)} · 打印 {escape_html(now)}</p>
</header>
<div class="doc-body">{payload.get("strategy_html") or ""}</div>
<footer class="doc-foot">文档:{escape_html(str(source))}{(" · " + escape_html(version)) if version else ""}</footer>
</article>"""
    else:
        raise KeyError(part)

    return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{escape_html(page_title)}</title>
<style>{css}</style>
</head>
<body>
{body}
{_print_auto_script()}
</body>
</html>"""


def build_export_html(exchange_key: str) -> str:
    payload = load_strategy_payload(exchange_key)
    key = payload["exchange_key"]
    label = payload["label"]
    title = payload["title"]
    version = payload.get("version") or ""
    checklist = payload.get("checklist") or {}
    now = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M")
    checklist_html = _checklist_html(checklist)

    return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{escape_html(label)} · 策略说明</title>
<style>
body{{font-family:system-ui,sans-serif;margin:24px;color:#111;line-height:1.55}}
h1{{font-size:1.35rem;margin:0 0 8px}}
.meta{{color:#555;font-size:.85rem;margin-bottom:20px}}
.grid{{display:grid;grid-template-columns:1fr 1fr;gap:24px;align-items:start}}
.strategy-doc h2{{font-size:1.05rem;margin-top:1.2em}}
.strategy-doc table{{border-collapse:collapse;width:100%;font-size:.88rem}}
.strategy-doc th,.strategy-doc td{{border:1px solid #ccc;padding:6px 8px;text-align:left}}
.checklist ul{{list-style:none;padding:0;margin:0}}
.checklist li{{display:flex;align-items:flex-start;gap:8px;margin:8px 0}}
.box{{flex:0 0 auto;display:inline-block;width:.95em;height:.95em;margin-top:.2em;border:1.5px solid #333;border-radius:2px;background:#fff;box-sizing:border-box}}
.item-text{{flex:1 1 auto}}
@media print{{body{{margin:12mm}}}}
</style>
</head>
<body>
<h1>{escape_html(title)}</h1>
<div class="meta">{escape_html(label)} · {escape_html(version)} · 导出 {now}</div>
<div class="grid">
  <div class="strategy-doc">{payload.get("strategy_html") or ""}</div>
  <div class="checklist">{checklist_html}</div>
</div>
</body>
</html>"""


def escape_html(text: str) -> str:
    from html import escape

    return escape(text, quote=True)