# -*- coding: utf-8 -*- from __future__ import annotations import json import secrets import time from datetime import datetime, timedelta, timezone from functools import wraps from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer BJT = timezone(timedelta(hours=8)) SSO_SALT = "gold-monitor-device-web-sso" SSO_MAX_AGE = 120 # 秒;设备免密登录票据有效期 from flask import ( Flask, Response, flash, g, jsonify, redirect, render_template, request, send_from_directory, session, stream_with_context, url_for, ) from app.config import Config, RELEASES_DIR from app import db as database def login_required(view): @wraps(view) def wrapped(*args, **kwargs): if not session.get("user_id"): return redirect(url_for("login", next=request.path)) return view(*args, **kwargs) return wrapped def admin_required(view): @wraps(view) def wrapped(*args, **kwargs): if not session.get("user_id"): return redirect(url_for("login", next=request.path)) if session.get("role") != "admin": flash("需要管理员权限", "error") return redirect(url_for("dashboard")) return view(*args, **kwargs) return wrapped def current_user() -> dict: return { "id": session.get("user_id"), "username": session.get("username"), "role": session.get("role"), "account_id": session.get("account_id"), } def register_routes(app: Flask) -> None: @app.context_processor def inject_globals(): return { "app_name": "Gold Monitor", "public_base": Config.PUBLIC_BASE_URL, "current_user": current_user(), "is_admin": session.get("role") == "admin", } def _sso_serializer() -> URLSafeTimedSerializer: return URLSafeTimedSerializer(Config.SECRET_KEY, salt=SSO_SALT) def _establish_session(u: dict, account_id=None) -> None: session.clear() session["user_id"] = u["id"] session["username"] = u["username"] session["role"] = u["role"] # 设备免密:会话账户以激活绑定账户为准,便于只看本账户复盘 session["account_id"] = account_id if account_id is not None else u.get("account_id") session["user"] = u["username"] def _safe_next(raw: str) -> str: nxt = (raw or "").strip() or url_for("records_page") if not nxt.startswith("/") or nxt.startswith("//"): return url_for("records_page") return nxt @app.get("/login") def login(): if session.get("user_id"): return redirect(url_for("dashboard")) return render_template("login.html") @app.post("/login") def login_post(): user = (request.form.get("username") or "").strip() pwd = request.form.get("password") or "" u = database.verify_login(user, pwd) if not u: flash("用户名或密码错误", "error") return render_template("login.html"), 401 _establish_session(u) return redirect(request.args.get("next") or url_for("dashboard")) @app.get("/logout") def logout(): session.clear() return redirect(url_for("login")) @app.get("/auth/device-login") def device_login(): """设备 token / 短期票据 → 网页会话(免密进复盘)。""" nxt = _safe_next(request.args.get("next") or "") ticket = (request.args.get("ticket") or "").strip() token = (request.args.get("token") or "").strip() u = None account_id = None try: if ticket: data = _sso_serializer().loads(ticket, max_age=SSO_MAX_AGE) with database.get_db() as conn: u = database.get_user(conn, int(data["uid"])) account_id = data.get("aid") if u and int(u.get("enabled") or 0) != 1: u = None elif token: with database.get_db() as conn: act = database.auth_token(conn, token) if not act: flash("设备未激活或 token 无效", "error") return redirect(url_for("login")) u = database.resolve_web_user_for_activation(conn, act) account_id = act.get("account_id") else: flash("缺少登录凭证", "error") return redirect(url_for("login")) except SignatureExpired: flash("登录链接已过期,请从客户端重新打开复盘", "error") return redirect(url_for("login")) except (BadSignature, KeyError, TypeError, ValueError): flash("登录凭证无效", "error") return redirect(url_for("login")) if not u: flash("该激活账户未绑定网页用户,请管理员在「用户」中创建并绑定账户", "error") return redirect(url_for("login")) _establish_session(u, account_id=account_id) return redirect(nxt) @app.get("/settings") @login_required def settings_page(): return render_template("settings.html") @app.post("/settings/profile") @login_required def settings_profile(): old_pwd = request.form.get("old_password") or "" new_username = (request.form.get("username") or "").strip() new_pwd = request.form.get("new_password") or "" new_pwd2 = request.form.get("new_password2") or "" if new_pwd or new_pwd2: if new_pwd != new_pwd2: flash("两次输入的新密码不一致", "error") return redirect(url_for("settings_page")) else: new_pwd = None try: prev = session.get("username") or "" updated = database.change_user_credentials( int(session["user_id"]), old_pwd, new_username=new_username, new_password=new_pwd, ) session["username"] = updated["username"] session["user"] = updated["username"] parts = [] if updated["username"] != prev: parts.append("用户名") if new_pwd: parts.append("密码") flash(("、".join(parts) or "资料") + "已更新", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("settings_page")) @app.post("/settings/password") @login_required def settings_password(): """兼容旧表单:仅改密码。""" old_pwd = request.form.get("old_password") or "" new_pwd = request.form.get("new_password") or "" new_pwd2 = request.form.get("new_password2") or "" if new_pwd != new_pwd2: flash("两次输入的新密码不一致", "error") return redirect(url_for("settings_page")) try: database.change_user_password(int(session["user_id"]), old_pwd, new_pwd) flash("密码已更新,请牢记新密码", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("settings_page")) # ---------- 用户管理(仅管理员)---------- @app.get("/users") @admin_required def users_page(): with database.get_db() as conn: users = database.list_users(conn) accounts = database.list_accounts(conn) # 未被占用的账户(用于新建普通用户) used = {u["account_id"] for u in users if u.get("account_id")} free_accounts = [a for a in accounts if a["id"] not in used] return render_template( "users.html", users=users, accounts=accounts, free_accounts=free_accounts, ) @app.post("/users") @admin_required def users_create(): username = (request.form.get("username") or "").strip() password = request.form.get("password") or "" role = (request.form.get("role") or "user").strip() account_raw = request.form.get("account_id") or "" account_id = int(account_raw) if account_raw.strip().isdigit() else None try: with database.get_db() as conn: database.create_user(conn, username, password, role, account_id) flash("用户已创建", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("users_page")) def _fmt_ts(value) -> str: if value is None or value == "": return "—" try: return datetime.fromtimestamp(float(value), BJT).strftime("%Y-%m-%d %H:%M:%S") except (TypeError, ValueError, OSError): return str(value) def _fmt_hold(open_ts, close_ts) -> str: """持仓时长:平仓−开仓;未平仓则按当前时间估算。""" if open_ts is None or open_ts == "": return "—" try: start = float(open_ts) end = float(close_ts) if close_ts not in (None, "") else time.time() except (TypeError, ValueError): return "—" secs = max(0, int(end - start)) days, rem = divmod(secs, 86400) hours, rem = divmod(rem, 3600) mins, sec = divmod(rem, 60) if days > 0: return f"{days}天{hours}小时{mins}分" if hours > 0: return f"{hours}小时{mins}分" if mins > 0: return f"{mins}分{sec}秒" return f"{sec}秒" def _annotate_trade_times(t: dict) -> None: t["open_time_fmt"] = _fmt_ts(t.get("open_time")) t["close_time_fmt"] = _fmt_ts(t.get("close_time")) closed = t.get("close_time") not in (None, "") t["hold_duration"] = _fmt_hold(t.get("open_time"), t.get("close_time")) if not closed and t.get("open_time"): t["hold_duration"] = t["hold_duration"] + "(持仓中)" t["review_issues_list"] = database.parse_review_issues(t.get("review_issues")) def _dashboard_payload(u: dict) -> dict: # 总览只看当前持仓(未平仓),历史成交在「交易复盘」 with database.get_db() as conn: if u["role"] == "admin": accounts = database.list_accounts(conn) trades = database.list_trades(conn, 100, open_only=True) else: accounts = [ a for a in database.list_accounts(conn) if a["id"] == u.get("account_id") ] trades = database.list_trades( conn, 100, account_id=u.get("account_id"), open_only=True ) machines = database.list_machines_for_user(conn, u) online_machines = [m for m in machines if m.get("online")] for m in machines: m["last_seen_fmt"] = _fmt_ts(m.get("last_seen_at")) for t in trades: t["time_fmt"] = _fmt_ts(t.get("open_time") or t.get("created_at")) return { "accounts": accounts, "machines": machines, "online_machines": online_machines, "online_count": len(online_machines), "machine_count": len(machines), "trades": trades, } @app.get("/") @login_required def dashboard(): data = _dashboard_payload(current_user()) return render_template("dashboard.html", **data) def _dashboard_live_dict(u: dict) -> dict: data = _dashboard_payload(u) return { "ok": True, "ts": database.now(), "online_count": data["online_count"], "machine_count": data["machine_count"], "online_machines": [ { "id": m["id"], "name": m.get("name") or f"#{m['id']}", "last_seen_at": m.get("last_seen_at"), "last_seen_fmt": _fmt_ts(m.get("last_seen_at")), "online": True, } for m in data["online_machines"] ], "trades": [ { "id": t["id"], "time_fmt": t.get("time_fmt"), "symbol": t.get("symbol"), "side": t.get("side"), "volume": t.get("volume"), "profit": t.get("profit"), "close_time": t.get("close_time"), "account_name": t.get("account_name"), "machine_name": t.get("machine_name"), "ticket": t.get("ticket"), } for t in data["trades"] ], } @app.get("/api/dashboard/live") @login_required def dashboard_live(): """总览快照(兼容轮询)。""" return jsonify(_dashboard_live_dict(current_user())) @app.get("/api/dashboard/stream") @login_required def dashboard_stream(): """秒级 SSE 推送:在线机器 + 实时成交。""" u = current_user() @stream_with_context def event_stream(): # 立即推一帧,后续约每秒一帧 while True: try: payload = json.dumps(_dashboard_live_dict(u), ensure_ascii=False) yield f"data: {payload}\n\n" except GeneratorExit: return except Exception: yield 'event: error\ndata: {"ok":false}\n\n' time.sleep(1) resp = Response(event_stream(), mimetype="text/event-stream") resp.headers["Cache-Control"] = "no-cache" resp.headers["X-Accel-Buffering"] = "no" resp.headers["Connection"] = "keep-alive" return resp # ---------- 账户(仅管理员创建)---------- @app.get("/accounts") @login_required def accounts_page(): u = current_user() with database.get_db() as conn: items = database.list_accounts(conn) if u["role"] != "admin": items = [a for a in items if a["id"] == u.get("account_id")] return render_template("accounts.html", items=items, can_create=(u["role"] == "admin")) @app.post("/accounts") @admin_required def accounts_create(): name = (request.form.get("name") or "").strip() if not name: flash("名称不能为空", "error") return redirect(url_for("accounts_page")) with database.get_db() as conn: conn.execute( """INSERT INTO accounts(name, broker, login_id, account_type, note, enabled, created_at) VALUES(?,?,?,?,?,?,?)""", ( name, (request.form.get("broker") or "").strip(), (request.form.get("login_id") or "").strip(), (request.form.get("account_type") or "demo").strip(), (request.form.get("note") or "").strip(), 1, database.now(), ), ) flash("账户已创建(仅管理员可创建)", "ok") return redirect(url_for("accounts_page")) @app.post("/accounts//note") @login_required def accounts_note(aid: int): u = current_user() if not database.user_can_access_account(u, aid): flash("无权修改该账户", "error") return redirect(url_for("accounts_page")) note = (request.form.get("note") or "").strip() with database.get_db() as conn: conn.execute("UPDATE accounts SET note=? WHERE id=?", (note, aid)) flash("标注已保存", "ok") return redirect(url_for("accounts_page")) @app.post("/accounts//delete") @admin_required def accounts_delete(aid: int): try: with database.get_db() as conn: database.delete_account(conn, aid) flash("账户已删除", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("accounts_page")) @app.get("/machines") @login_required def machines_page(): u = current_user() with database.get_db() as conn: items = database.list_machines_for_user(conn, u) for m in items: m["last_seen_fmt"] = _fmt_ts(m.get("last_seen_at")) m["bound"] = bool(m.get("device_fingerprint")) return render_template("machines.html", items=items, is_admin=(u["role"] == "admin")) @app.post("/machines") @admin_required def machines_create(): name = (request.form.get("name") or "").strip() or "新机器" with database.get_db() as conn: conn.execute( """INSERT INTO machines(name, device_fingerprint, note, enabled, created_at) VALUES(?,?,?,?,?)""", (name, "", (request.form.get("note") or "").strip(), 1, database.now()), ) flash("机器已添加(预登记,激活后绑定指纹)", "ok") return redirect(url_for("machines_page")) @app.post("/machines//note") @login_required def machines_note(mid: int): note = (request.form.get("note") or "").strip() name = (request.form.get("name") or "").strip() with database.get_db() as conn: if name: conn.execute("UPDATE machines SET note=?, name=? WHERE id=?", (note, name, mid)) else: conn.execute("UPDATE machines SET note=? WHERE id=?", (note, mid)) flash("机器标注已保存", "ok") return redirect(url_for("machines_page")) @app.post("/machines//delete") @admin_required def machines_delete(mid: int): try: with database.get_db() as conn: database.delete_machine(conn, mid) flash("机器已删除", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("machines_page")) # ---------- 下载 / 激活(管理员与用户都可发码;用户仅自己账户)---------- @app.get("/download") @login_required def download_page(): u = current_user() with database.get_db() as conn: all_accounts = database.list_accounts(conn) if u["role"] == "admin": accounts = all_accounts else: accounts = [a for a in all_accounts if a["id"] == u.get("account_id")] machines = database.list_machines(conn) if u["role"] == "admin": acts = conn.execute( """ SELECT act.*, a.name AS account_name, m.name AS machine_name, m.last_seen_at AS machine_last_seen FROM activations act LEFT JOIN accounts a ON a.id = act.account_id LEFT JOIN machines m ON m.id = act.machine_id ORDER BY act.id DESC LIMIT 50 """ ).fetchall() else: acts = conn.execute( """ SELECT act.*, a.name AS account_name, m.name AS machine_name, m.last_seen_at AS machine_last_seen FROM activations act LEFT JOIN accounts a ON a.id = act.account_id LEFT JOIN machines m ON m.id = act.machine_id WHERE act.account_id=? ORDER BY act.id DESC LIMIT 50 """, (u.get("account_id"),), ).fetchall() activations = [] now_ts = database.now() online_sec = database.ONLINE_SECONDS for r in acts: item = database.row_to_dict(r) last = item.get("machine_last_seen") if item.get("status") == "active" and last and (now_ts - float(last)) <= online_sec: item["online_label"] = "在线" item["online"] = True else: item["online_label"] = "离线" item["online"] = False try: item["created_at_fmt"] = datetime.fromtimestamp(float(item["created_at"])).strftime( "%Y-%m-%d %H:%M" ) except (TypeError, ValueError, OSError): item["created_at_fmt"] = str(item.get("created_at") or "—") activations.append(item) exe_name = "黄金市价辅助.exe" exe_path = RELEASES_DIR / exe_name return render_template( "download.html", accounts=accounts, machines=machines, activations=activations, exe_ready=exe_path.is_file(), exe_name=exe_name, ) @app.post("/download/activation") @login_required def download_create_activation(): u = current_user() account_id = int(request.form.get("account_id") or 0) machine_raw = request.form.get("machine_id") or "" machine_id = int(machine_raw) if machine_raw.strip().isdigit() else None if not database.user_can_access_account(u, account_id): flash("只能为自己的账户发放激活码", "error") return redirect(url_for("download_page")) try: with database.get_db() as conn: code = database.create_activation(conn, account_id, machine_id, u.get("id")) flash(f"已生成激活码:{code}", "ok") except ValueError as exc: flash(str(exc), "error") return redirect(url_for("download_page")) @app.get("/download/file") @login_required def download_file(): name = "黄金市价辅助.exe" path = RELEASES_DIR / name if not path.is_file(): readme = RELEASES_DIR / "README_PLACE_EXE_HERE.txt" if not readme.is_file(): readme.write_text( "请将打包好的通用辅助 exe 放到本目录,命名为:黄金市价辅助.exe\n", encoding="utf-8", ) return send_from_directory(RELEASES_DIR, readme.name, as_attachment=True) return send_from_directory(RELEASES_DIR, name, as_attachment=True) @app.post("/api/v1/activate") def api_activate(): data = request.get_json(silent=True) or {} code = (data.get("code") or "").strip() fingerprint = (data.get("fingerprint") or secrets.token_hex(8)).strip() machine_name = (data.get("machine_name") or "").strip() if not code: return jsonify({"ok": False, "error": "缺少激活码"}), 400 try: with database.get_db() as conn: result = database.activate_device(conn, code, fingerprint, machine_name) return jsonify({"ok": True, **result}) except ValueError as exc: return jsonify({"ok": False, "error": str(exc)}), 400 @app.get("/api/v1/ping") def api_ping(): token = request.headers.get("X-Device-Token") or request.args.get("token") or "" with database.get_db() as conn: act = database.auth_token(conn, token) if not act: return jsonify({"ok": False, "error": "未激活"}), 401 if act.get("machine_id"): conn.execute( "UPDATE machines SET last_seen_at=? WHERE id=?", (database.now(), act["machine_id"]), ) return jsonify({"ok": True, "server_time": database.now()}) @app.post("/api/v1/web-sso") def api_web_sso(): """用设备 token 换短期免密登录链接(打开复盘用)。""" token = request.headers.get("X-Device-Token") or "" data = request.get_json(silent=True) or {} nxt = _safe_next(data.get("next") or "/records") with database.get_db() as conn: act = database.auth_token(conn, token) if not act: return jsonify({"ok": False, "error": "未激活"}), 401 u = database.resolve_web_user_for_activation(conn, act) if not u: return jsonify({ "ok": False, "error": "该激活账户未绑定网页用户,请管理员在「用户」中创建并绑定账户", }), 403 if act.get("machine_id"): conn.execute( "UPDATE machines SET last_seen_at=? WHERE id=?", (database.now(), act["machine_id"]), ) ticket = _sso_serializer().dumps({ "uid": int(u["id"]), "aid": act.get("account_id"), "mid": act.get("machine_id"), }) path = url_for("device_login", ticket=ticket, next=nxt) return jsonify({ "ok": True, "ticket": ticket, "path": path, "expires_in": SSO_MAX_AGE, }) @app.post("/api/v1/trades") def api_trades_upsert(): token = request.headers.get("X-Device-Token") or "" data = request.get_json(silent=True) or {} with database.get_db() as conn: act = database.auth_token(conn, token) if not act: return jsonify({"ok": False, "error": "未激活"}), 401 if not act.get("account_id"): return jsonify({"ok": False, "error": "激活码未绑定账户"}), 400 snapshot = data.get("snapshot") or {} ticket = str(data.get("ticket") or "").strip() symbol = (data.get("symbol") or "XAUUSD").upper() side = (data.get("side") or "").lower() volume = round(float(data.get("volume") or 0), 4) open_price = round(float(data.get("open_price") or 0), 3) close_price = ( round(float(data.get("close_price") or 0), 3) if data.get("close_price") is not None else None ) open_time = data.get("open_time") close_time = data.get("close_time") if data.get("profit") is None: profit = None else: profit = round(float(data.get("profit") or 0), 2) sl = data.get("sl") tp = data.get("tp") if sl is not None and sl != "": try: sl = round(float(sl), 3) except (TypeError, ValueError): sl = None else: sl = None if tp is not None and tp != "": try: tp = round(float(tp), 3) except (TypeError, ValueError): tp = None else: tp = None entry_style = (data.get("entry_style") or "").strip() review_note = (data.get("review_note") or "").strip() snap_json = json.dumps(snapshot, ensure_ascii=False) existing = None if ticket: # 优先更新仍未平仓的那条,避免历史补报插到已平仓副本上 existing = conn.execute( """ SELECT id, entry_style, review_note, open_time, close_time, close_price FROM trades WHERE account_id=? AND ticket=? ORDER BY CASE WHEN close_time IS NULL OR close_time='' THEN 0 ELSE 1 END, id DESC LIMIT 1 """, (act["account_id"], ticket), ).fetchone() if existing: # 浮盈更新不覆盖已有平仓;空备注不覆盖已标注 keep_close = existing["close_time"] not in (None, "") and close_time in (None, "") new_close_time = existing["close_time"] if keep_close else close_time new_close_price = existing["close_price"] if keep_close else close_price new_entry = entry_style or (existing["entry_style"] or "") new_note = review_note or (existing["review_note"] or "") new_open_time = open_time if open_time is not None else existing["open_time"] if keep_close: # 已平仓单不再被浮盈推送改写盈亏 profit = conn.execute( "SELECT profit FROM trades WHERE id=?", (existing["id"],) ).fetchone()["profit"] conn.execute( """ UPDATE trades SET symbol=?, side=?, volume=?, open_price=?, close_price=?, open_time=?, close_time=?, profit=?, sl=?, tp=?, machine_id=?, entry_style=?, review_note=?, snapshot_json=? WHERE id=? """, ( symbol, side, volume, open_price, new_close_price, new_open_time, new_close_time, profit, sl, tp, act["machine_id"], new_entry, new_note, snap_json, existing["id"], ), ) tid = int(existing["id"]) else: conn.execute( """ INSERT INTO trades( ticket, symbol, side, volume, open_price, close_price, open_time, close_time, profit, sl, tp, account_id, machine_id, entry_style, review_note, snapshot_json, created_at ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( ticket, symbol, side, volume, open_price, close_price, open_time, close_time, profit, sl, tp, act["account_id"], act["machine_id"], entry_style, review_note, snap_json, database.now(), ), ) tid = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0]) # 同一账户+单号只保留一条有效记录: # - 上报持仓:关掉其它未平仓副本 # - 上报平仓:其它同单号副本盈亏清空(防 history 重复插入双计) if ticket: dup_snap = json.dumps( {"event": "dedupe_closed", "kept_id": tid}, ensure_ascii=False ) if close_time in (None, ""): conn.execute( """ UPDATE trades SET close_time=?, profit=NULL, snapshot_json=? WHERE account_id=? AND ticket=? AND id!=? AND (close_time IS NULL OR close_time='') """, (database.now(), dup_snap, act["account_id"], ticket, tid), ) else: conn.execute( """ UPDATE trades SET profit=NULL, snapshot_json=? WHERE account_id=? AND ticket=? AND id!=? """, (dup_snap, act["account_id"], ticket, tid), ) return jsonify({"ok": True, "id": tid}) @app.post("/api/v1/trades/reconcile") def api_trades_reconcile(): """客户端上报当前仍持仓的 ticket;库中其它未平仓记录结束「持仓中」。 按账户对账(不按品种过滤),避免 XAUUSD/XAUUSDm 不一致导致幽灵持仓。 同一 ticket 多条未平仓时只保留 id 最大的一条。 """ token = request.headers.get("X-Device-Token") or "" data = request.get_json(silent=True) or {} open_tickets = {str(t).strip() for t in (data.get("open_tickets") or []) if str(t).strip()} with database.get_db() as conn: act = database.auth_token(conn, token) if not act: return jsonify({"ok": False, "error": "未激活"}), 401 if not act.get("account_id"): return jsonify({"ok": False, "error": "激活码未绑定账户"}), 400 rows = conn.execute( """ SELECT id, ticket, profit, snapshot_json FROM trades WHERE account_id=? AND (close_time IS NULL OR close_time='') ORDER BY id DESC """, (act["account_id"],), ).fetchall() closed_n = 0 now_ts = database.now() seen_open: set[str] = set() for r in rows: tid = str(r["ticket"] or "") keep = tid in open_tickets and tid not in seen_open if keep: seen_open.add(tid) continue try: snap = json.loads(r["snapshot_json"] or "{}") except json.JSONDecodeError: snap = {} snap["event"] = "reconcile_closed" snap["reconciled_at"] = now_ts # 假平仓不写入冻结浮盈,避免脏数据进统计 conn.execute( """ UPDATE trades SET close_time=?, profit=NULL, snapshot_json=? WHERE id=? """, (now_ts, json.dumps(snap, ensure_ascii=False), r["id"]), ) closed_n += 1 return jsonify({"ok": True, "closed": closed_n}) @app.post("/api/v1/bars") def api_bars_upload(): token = request.headers.get("X-Device-Token") or "" data = request.get_json(silent=True) or {} with database.get_db() as conn: if not database.auth_token(conn, token): return jsonify({"ok": False, "error": "未激活"}), 401 symbol = (data.get("symbol") or "XAUUSD").upper() timeframe = (data.get("timeframe") or "M1").upper() bars = data.get("bars") or [] n = database.upsert_bars(conn, symbol, timeframe, bars) return jsonify({"ok": True, "count": n}) @app.get("/api/v1/bars") @login_required def api_bars_get(): symbol = (request.args.get("symbol") or "XAUUSD").upper() timeframe = (request.args.get("timeframe") or "M15").upper() start_ts = float(request.args.get("start") or 0) end_ts = float(request.args.get("end") or database.now()) with database.get_db() as conn: bars = database.get_bars(conn, symbol, timeframe, start_ts, end_ts) return jsonify({"ok": True, "bars": bars}) def _resolve_machine_filter(u: dict, machine_raw: str): machine_id = int(machine_raw) if machine_raw.isdigit() else None with database.get_db() as conn: machines = database.list_machines_for_user(conn, u) allowed_ids = {int(m["id"]) for m in machines} if machine_id is not None and machine_id not in allowed_ids: machine_id = None return machines, machine_id @app.get("/records") @login_required def records_page(): u = current_user() machine_raw = (request.args.get("machine_id") or "").strip() page_raw = (request.args.get("page") or "1").strip() page = int(page_raw) if page_raw.isdigit() and int(page_raw) >= 1 else 1 per_page = 10 account_id = None if u["role"] == "admin" else u.get("account_id") machines, machine_id = _resolve_machine_filter(u, machine_raw) with database.get_db() as conn: database.scrub_synthetic_trade_profits(conn) database.scrub_duplicate_ticket_profits(conn) total = database.count_trades( conn, account_id=account_id, machine_id=machine_id, hide_synthetic=True, ) total_pages = max(1, (total + per_page - 1) // per_page) if page > total_pages: page = total_pages offset = (page - 1) * per_page trades = database.list_trades( conn, limit=per_page, account_id=account_id, machine_id=machine_id, offset=offset, hide_synthetic=True, ) for t in trades: _annotate_trade_times(t) return render_template( "records.html", trades=trades, machines=machines, selected_machine_id=machine_id, page=page, per_page=per_page, total=total, total_pages=total_pages, ) @app.get("/api/records/fingerprint") @login_required def records_fingerprint(): """复盘列表/统计页轮询:成交或标注变化时前端自动刷新。""" u = current_user() machine_raw = (request.args.get("machine_id") or "").strip() account_id = None if u["role"] == "admin" else u.get("account_id") _machines, machine_id = _resolve_machine_filter(u, machine_raw) with database.get_db() as conn: fp = database.trades_fingerprint( conn, account_id=account_id, machine_id=machine_id ) return jsonify({"ok": True, "fp": fp, "ts": database.now()}) @app.get("/records/stats") @login_required def records_stats_page(): u = current_user() machine_raw = (request.args.get("machine_id") or "").strip() period = (request.args.get("period") or "today").strip().lower() if period not in ("today", "all"): period = "today" day = (request.args.get("day") or "").strip() or None account_id = None if u["role"] == "admin" else u.get("account_id") machines, machine_id = _resolve_machine_filter(u, machine_raw) with database.get_db() as conn: # 清掉假平仓盈亏 + 同单号重复平仓双计 database.scrub_synthetic_trade_profits(conn) database.scrub_duplicate_ticket_profits(conn) stats = database.trade_stats( conn, account_id=account_id, machine_id=machine_id, day=day, period=period, ) return render_template( "records_stats.html", stats=stats, machines=machines, selected_machine_id=machine_id, selected_period=period, selected_day=stats.get("day") or day or "", ) @app.get("/records/") @login_required def record_detail(tid: int): u = current_user() with database.get_db() as conn: trade = database.get_trade(conn, tid) if not trade or not database.user_can_access_account(u, trade.get("account_id")): flash("记录不存在或无权查看", "error") return redirect(url_for("records_page")) _annotate_trade_times(trade) # 按目标根数取足够时间窗:M1=500, M5=200, M15=100, H1=50 open_ts = float( trade.get("open_time") or trade.get("created_at") or database.now() ) close_raw = trade.get("close_time") close_ts = float(close_raw) if close_raw not in (None, "") else open_ts # 秒/根 × 目标根数,再略留余量 tf_span = { "M1": 500 * 60, "M5": 200 * 5 * 60, "M15": 100 * 15 * 60, "H1": 50 * 3600, } charts = {} for tf in ("M1", "M5", "M15", "H1"): span = tf_span[tf] mid = (min(open_ts, close_ts) + max(open_ts, close_ts)) / 2.0 charts[tf] = database.get_bars( conn, trade["symbol"], tf, mid - span / 2.0, mid + span / 2.0, ) trade["review_issues_list"] = database.parse_review_issues( trade.get("review_issues") ) return render_template( "record_detail.html", trade=trade, charts=charts, entry_styles=database.ENTRY_STYLES, entry_patterns=database.ENTRY_PATTERN_TYPES, review_issues=database.REVIEW_ISSUES, review_issues_max=database.REVIEW_ISSUES_MAX, ) @app.post("/records//annotate") @login_required def record_annotate(tid: int): u = current_user() entry_style = (request.form.get("entry_style") or "").strip() note = (request.form.get("review_note") or "").strip() raw_issues = request.form.getlist("review_issues") issues = database.parse_review_issues(raw_issues) if len(issues) > database.REVIEW_ISSUES_MAX: flash(f"主要问题最多选 {database.REVIEW_ISSUES_MAX} 个", "error") return redirect(url_for("record_detail", tid=tid)) issues_s = database.serialize_review_issues(issues) allowed = set(database.ENTRY_STYLES) with database.get_db() as conn: trade = database.get_trade(conn, tid) if not trade or not database.user_can_access_account(u, trade.get("account_id")): flash("无权标注该记录", "error") return redirect(url_for("records_page")) # 允许清空;或保留已有旧标注值;新选必须在标准列表内 prev = (trade.get("entry_style") or "").strip() if entry_style and entry_style not in allowed and entry_style != prev: flash("入场形式不在可选列表中", "error") return redirect(url_for("record_detail", tid=tid)) conn.execute( "UPDATE trades SET entry_style=?, review_issues=?, review_note=? WHERE id=?", (entry_style, issues_s, note, tid), ) flash("复盘标注已保存", "ok") return redirect(url_for("record_detail", tid=tid)) @app.get("/health") def health(): return jsonify({"ok": True, "service": "gold_monitor", "port": Config.PORT})