# -*- coding: utf-8 -*- """页面的本机预览服务 (只给开发机看样式用, 不进容器、不连库、不进代码指纹)。 用途: 改 app/web/static/index.html 之后, 不部署就能在浏览器里看效果。它只做几件事: 静态托管页面; 假登录 (交易员加系统管理员两个角色); 给顶栏、持仓表、建仓候选栏、个股详情抽屉、 评析报告抽屉、运维视图「组合操作」页喂固定的假数据。其余 /api/* 一律回「预览无后端」, 页面照常挂载, 顶部会有一条错误提示, 不影响看样式。 怎么跑 (开发机, 需要一个装了 fastapi 与 uvicorn 的 Python, 用项目 requirements.txt 建的虚拟环境即可): ~/venvs/pms/bin/python docs/dev_preview/mock_preview.py 浏览器开 http://127.0.0.1:38199/ 看主页面 浏览器开 http://127.0.0.1:38199/#detail=300627.SZ 直达个股详情抽屉 两个坑: 一, 仓库若放在 macOS 的「文稿」目录下, 由别的程序 (例如 Claude 桌面版的预览面板) 拉起的进程可能没有 读这个目录的权限, 表现为页面 404 或读不到文件。办法是把 app/web/static 整个拷到一个临时目录, 再用环境变量 PMS_PREVIEW_STATIC 指过去: cp -R app/web/static /tmp/pms_static && PMS_PREVIEW_STATIC=/tmp/pms_static python docs/dev_preview/mock_preview.py 改了页面要重新拷一次。 二, 假数据的键要跟着接口走。交易员视图的持仓表只显示 total_qty 大于零的行; 顶栏读 /health、/api/overview、 /api/tech/status、/api/macro/status、/api/dispatch-mode; 运维视图「组合操作」页读 /api/params。 页面新读了哪个接口, 就在这里补一个同形状的假应答。 本文件放在 docs/ 下是有意的: 代码指纹只覆盖 app/、scripts/、config/ 三个目录的 .py, 放这里不会让 服务器上的 make stale 报「镜像与工作树不一致」。 """ import os from fastapi import FastAPI from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.environ.get("PMS_PREVIEW_STATIC") or os.path.abspath(os.path.join(HERE, "..", "..", "app", "web", "static")) SAMPLE_REVIEW = os.path.join(HERE, "sample_review.md") app = FastAPI() app.mount("/static", StaticFiles(directory=ROOT), name="static") @app.get("/") def index(): return FileResponse(os.path.join(ROOT, "index.html")) @app.get("/api/me") def me(): return {"ok": True, "user": {"phone": "", "username": "预览"}, "roles": ["交易员", "系统管理员"]} RESEARCH = { "ts_code": "300750.SZ", "company": {"overall": "好", "valuation": "便宜", "period": "2026Q2", "line": "质地好（回报好、盈余中、成长好），估值便宜，置信中", "invalidation": "下一期财报看M24净利率、M26经营现金流对净利润、M63收入同比、M64净利润同比", "doubt_hard": False, "report_url": "https://example.invalid/report/300750", "text": ""}, "fund": {"stance": "看多", "fact": "质地好", "mark": None, "priced_in": False, "valuation": "便宜", "overall": "好", "no_read_why": None}, "tech": {"stance": "看空", "strength": "强", "phase": "趋势空", "confirm": None, "sar_side": "空", "sar_value": 344.446, "sar_flip_days": 18, "choppy": False, "reason": "SAR 空且多空布林线未在多头区：下行", "data_date": 20260911, "latest": {"data_date": 20260911, "boll": {"upper": 352.1, "mid": 338.0, "lower": 323.9, "bw_pct": 8.3, "pos": 0.12, "state": "贴下轨", "squeeze": False}, "bbi": {"bbi": 341.2, "pos": -0.6, "state": "空头区", "dist_pct": -1.2}, "sar": {"value": 344.446, "side": "空", "flip_days": 18, "dist_pct": 2.1}, "quality": "OK", "reanchored": True}, "flips": 1, "bars": 25}, "timing": {"stance": "中性", "nightly": "WATCH", "trade_date": "20260911", "flip_at": None}, "consensus": {"direction": "中性", "votes": {"fund": "看多", "tech": "看空", "timing": "中性"}, "strength": None, "reason": "基本面看多、技术面看空、择时中性 → 中性", "route": "观察", "route_reason": "等技术面转向"}, "sources": {"company": "ok", "fund": "ok", "tech": "ok", "timing": "ok", "consensus": "ok"}, } @app.get("/api/research/{code}") def research(code: str): return {"ok": True, "data": {**RESEARCH, "ts_code": code}} @app.get("/api/upstream/signals/{code}") def signals(code: str): return {"ok": True, "data": { "ok": True, "ts_code": code, "timeline": [ {"time": "10:53", "src": "alert", "src_label": "盘中告警", "cat": "money_flow", "cat_label": "资金流强度", "direction": "看涨", "level": "INFO", "reason": "money_flow_in_intensity", "stale": True, "dup": 1, "ts_from": "trigger"}, ], "snapshot": {"metrics": [{"direction": "看涨", "window_net": 432000000.0, "z_dd": 6.143}], "mr": {}}, "sources": {"intraday": {"ok": True}, "alerts": {"ok": True}, "sell_actions": {"ok": True}, "metrics": {"ok": True}, "mr": {"ok": True}}, "truncated": {}, }} @app.get("/api/ledger") def ledger(ts_code: str = "", limit: int = 40): import datetime today = datetime.date.today().isoformat() return {"ok": True, "data": [ {"created_at": f"{today} 09:41:12", "action": "OPEN", "verdict": "REJECT", "arbiter": "system", "reason": "系统放弃：研究证据走弱（逻辑存疑），自动放弃；明日重新评估"}, ]} @app.get("/api/proposals") def proposals(status: str = "WAIT_USER", limit: int = 100): return {"ok": True, "data": []} # ---- 持仓表 (综合分析列三枚标签), 照 155 上 10:53 读回的三票造几行 ---- def _pos(code, name, qty, cost, price, dims, cushion="POSITIVE", cpct=0.052, frozen="NONE", target=None): mv = qty * price return { "ts_code": code, "name": name, "qty": qty, "total_qty": qty, "sellable_qty": qty, "avg_cost": cost, "price": price, "status": "HOLDING", "frozen_reason": frozen, "user_target_price": target, "price_ok": True, "market_value": mv, "pct_of_scale": mv / 2000000.0, "cushion_state": cushion, "cushion_pct": cpct, "neg_cushion_days": 0, "user_stop_price": None, "entry": {"why": "光模块需求上行"}, "logic_now": {"state": "INTACT", "text": "证据仍在"}, "tech": {"stance": dims["tech"] or "无读数", "phase": dims["tech_phase"]}, "dims": dims, "lots": 1, } POSITIONS = [ _pos("300627.SZ", "华测导航", 1200, 41.2, 43.5, {"fund": "看多", "fund_fact": "质地好", "tech": "看空", "tech_phase": "转空", "timing": "中性", "nightly": "WATCH"}), _pos("688281.SH", "华秦科技", 300, 158.0, 151.2, {"fund": "中性", "fund_fact": "质地中", "tech": "看空", "tech_phase": "转空", "timing": "中性", "nightly": "WATCH"}, cushion="NEGATIVE", cpct=-0.043), _pos("002463.SZ", "沪电股份", 200, 115.25, 118.9, {"fund": None, "fund_fact": "没有买方评析", "tech": "中性", "tech_phase": "收口等待", "timing": "中性", "nightly": "WATCH"}), _pos("002179.SZ", "中航光电", 500, 52.3, 55.1, {"fund": None, "fund_fact": "没有买方评析", "tech": None, "tech_phase": None, "timing": "看多", "nightly": "BUY"}, frozen="MANUAL", target=60.0), _pos("002709.SZ", "天赐材料", 800, 24.1, 22.7, {"fund": None, "fund_fact": "没有买方评析", "tech": "看空", "tech_phase": "趋势空", "timing": None, "nightly": "DROPPED"}, cushion="NEGATIVE", cpct=-0.058), _pos("600279.SH", "重庆港", 2000, 6.1, 6.2, {"fund": None, "fund_fact": "没有买方评析", "tech": None, "tech_phase": None, "timing": None, "nightly": None}), ] @app.get("/api/strategies") def strategies(include_archived: bool = False): return {"ok": True, "strategies": [ {"strategy_id": "STR_20260908_300627SZ_1", "ts_code": "300627.SZ", "type": "GRID", "status": "ACTIVE", "autonomy": "auto", "params": {"lower": 38.0, "upper": 46.0, "center": 42.0}, "state": {"filled_levels": {"1": 200, "2": 200}, "invested": 16800.0}, "archived_at": None, "buy_paused": None, "created_at": "2026-09-08 09:40:00"}, {"strategy_id": "STR_20260910_002463SZ_1", "ts_code": "002463.SZ", "type": "TRAIL", "status": "PAUSED", "autonomy": "auto", "params": {"trail_pct": 0.05}, "state": {"high_water": 121.4, "armed": False}, "archived_at": None, "buy_paused": None, "created_at": "2026-09-10 09:40:00"}, ], "enabled": True, "auto": {"enabled": True, "ran_today": True, "last_scan": {"at": "2026-09-15 09:40:12"}}} @app.get("/api/positions") def positions(): return {"ok": True, "positions": POSITIONS, "stock_params": {}, "summary": {"total_scale": 2000000.0, "market_value": sum(p["market_value"] for p in POSITIONS)}} # ---- 建仓候选栏: 计划行 + 只读处置快照 ---- PLAN_ROWS = [ {"rank": 1, "ts_code": "300308.SZ", "name": "中际旭创", "theme": "光模块与算力互联", "tier": "白名单"}, {"rank": 2, "ts_code": "300750.SZ", "name": "宁德时代", "theme": "动力电池", "tier": "白名单"}, {"rank": 3, "ts_code": "688981.SH", "name": "中芯国际", "theme": "晶圆代工与先进制程国产替代", "tier": "白名单"}, {"rank": 4, "ts_code": "002050.SZ", "name": "三花智控", "theme": "热管理与机器人执行器", "tier": "候补"}, {"rank": 5, "ts_code": "300660.SZ", "name": "江苏雷利", "theme": "微特电机", "tier": "候补"}, {"rank": 6, "ts_code": "603986.SH", "name": "兆易创新", "theme": "存储芯片", "tier": "候补"}, ] @app.get("/api/upstream/plan") def plan(limit: int = 50): return {"ok": True, "status": {"ok": True, "date": "2026-09-15", "plan_source": "snapshot", "gate_on": True, "age_tdays": 1, "fetched_at": __import__("time").time() - 3000, "heat_date": "2026-09-15"}, "rows": PLAN_ROWS, "candidates_raw": None} @app.get("/api/upstream/plan-changes") def plan_changes(limit: int = 50): return {"ok": True, "changes": {"ok": True, "added": [], "removed": []}, "log": {}} def _con(fund, overall, tech, phase, timing, direction, route): return {"fund": fund, "overall": overall, "tech": tech, "phase": phase, "timing": timing, "direction": direction, "route": route} @app.get("/api/open-scan") def open_scan(): return {"ok": True, "notes": ["名额 3/8", "可用资金 61.2 万"], "by_code": { "300308.SZ": {"disp": "would", "why": "名额与资金充足，本轮将建底仓", "consensus": _con("看多", "好", "看多", "趋势多", "看多", "看多", "放行")}, "300750.SZ": {"disp": "wait_tech", "why": "布林带仍在收口，等开口向上再评估", "consensus": _con("看多", "好", "中性", "收口等待", "中性", "看多", "交人")}, "688981.SH": {"disp": "wait_timing", "why": "三票看多但择时看空，等择时转多", "consensus": _con("看多", "好", "看多", "趋势多", "看空", "看多", "观察")}, "002050.SZ": {"disp": "wait_confirm", "why": "技术面已开口，等盘中站上均价且量比达标", "consensus": _con("中性", "中", "看多", "开口向上", "中性", "看多", "放行")}, "300660.SZ": {"disp": "sys_decline", "why": "系统放弃：研究证据走弱（逻辑存疑），明日重新评估", "consensus": _con("看空", "差", "中性", "收口等待", None, "中性", "跳过")}, }} # ---- 2026-09-18: 顶栏标签、运维视图组合操作页、评析报告抽屉要用的几路 ---- @app.get("/health") def health(): import datetime return {"ok": True, "version": "preview", "now": datetime.datetime.now().strftime("%H:%M:%S"), "trade_day": True} @app.get("/api/overview") def overview(): mv = sum(p["market_value"] for p in POSITIONS) return {"ok": True, "data": {"buy_halt": False, "exec_halt": False, "autonomy": "full", "scale": 2000000.0, "portfolio_mv": mv, "portfolio_pct": mv / 2000000.0, "portfolio_cap": 0.8, "cap_room": 2000000.0 * 0.8 - mv, "cash_source": "ws", "cash_avail": 612000.0, "cash_est": 600000.0, "names_count": len(POSITIONS), "max_names": 8, "float_pnl": 12345.0, "solid_names": 2, "neg_names": 2, "sector_ready": True}} @app.get("/api/params") def params(): def p(k, v, t="float", d=""): return {"key": k, "value": v, "type": t, "file_default": v, "source": "table", "desc": d, "updated_at": "2026-09-17 15:20:00"} return {"ok": True, "data": {"params": [ p("PMS_TOTAL_SCALE", 2000000, "int", "总操作规模"), p("PMS_PORTFOLIO_CAP", 0.8, "float", "总仓上限"), p("PMS_STOCK_CAP", 0.15, "float", "单股上限"), p("PMS_MAX_NAMES", 8, "int", "最多持仓"), p("PMS_CASH_RESERVE", 0.05, "float", "预留现金"), p("PMS_DCA_MAX_RATIO", 0.5, "float", "最多补到首批的"), p("PMS_DCA_TRIGGERS", "-0.08,-0.15", "str", "亏到多少开始评估"), p("PMS_DCA_DEEP_CONFIRM", True, "bool", "更深那档要确认"), p("PMS_FILL_MAX_LOSS", -0.05, "float", "回落补足的止损"), p("PMS_AUTONOMY", "full", "str", "自主档位"), ]}} @app.get("/api/tech/status") def tech_status(): return {"ok": True, "data": {"enabled": True, "data_date": 20260917, "map": {"fresh": True}}} @app.get("/api/macro/status") def macro_status(): return {"ok": True, "enabled": True, "autonomy": "propose_only", "params": {"hot_th": 1.5, "cold_th": -1.5, "exit_band": 0.3}, "signals": [{"key": "MZ_20260918", "label": "大盘冷热", "value": 0.4, "zone": "NEUTRAL", "ymd": "2026-09-18", "e_now": 0, "streak": 3, "history": [], "advice": None}]} @app.get("/api/dispatch-mode") def dispatch_mode(): return {"ok": True, "data": {"mode": "ws", "shadow": False}} @app.get("/api/research/{code}/report") def research_report(code: str): md = open(SAMPLE_REVIEW, encoding="utf-8").read() if code == "600279.SH": return {"ok": False, "ts_code": code, "url": None, "source": "none", "markdown": "", "cached": False, "fetched_at": None, "error": "这只票没有个股深度评析报告的链接（不在持仓也不在当日选股计划里）"} return {"ok": True, "ts_code": code, "url": "http://192.168.16.178:8000/company/%s/review" % code, "source": "held", "markdown": md, "cached": False, "fetched_at": "2026-09-18 11:50:00", "error": None} @app.api_route("/api/{path:path}", methods=["GET", "POST"]) def catchall(path: str): return JSONResponse({"ok": False, "error": "预览无后端"}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=38199)