"""中控 AI：四户数据聚合为结构化上下文。""" from __future__ import annotations import hashlib import json import os from datetime import datetime, timedelta from typing import Any, Optional import httpx from hub_ai.config import FUND_HISTORY_DAYS, hub_agent_timeout, hub_flask_timeout, trading_day_reset_hour from hub_ai.fund_history import format_fund_history_text, get_fund_history, record_fund_snapshot from hub_trades_lib import current_trading_day, summarize_trades def _hub_token() -> str: return (os.getenv("HUB_BRIDGE_TOKEN") or os.getenv("CONTROL_TOKEN") or "").strip() def _hub_headers() -> dict[str, str]: tok = _hub_token() return {"X-Hub-Token": tok} if tok else {} def _agent_headers() -> dict[str, str]: tok = (os.getenv("CONTROL_TOKEN") or os.getenv("HUB_BRIDGE_TOKEN") or "").strip() return {"X-Control-Token": tok} if tok else {} def _safe_float(v: Any) -> Optional[float]: try: if v is None or v == "": return None return float(v) except (TypeError, ValueError): return None def _position_contracts(p: dict) -> float: for key in ("contracts", "contracts_signed", "size"): v = p.get(key) try: if v is not None and v != "": return float(v) except (TypeError, ValueError): continue return 0.0 def _filter_open_positions(positions: list) -> list[dict]: out: list[dict] = [] for p in positions or []: if not isinstance(p, dict): continue if abs(_position_contracts(p)) < 1e-12: continue out.append(p) return out def _account_open_position_count(ac: dict) -> int: return len(_filter_open_positions(ac.get("positions") or [])) def _monitor_counts(ac: dict) -> dict[str, int]: mon = ac.get("monitor_lines") or {} return { "trends": len(mon.get("trends") or []), "rolls": len(mon.get("rolls") or []), "keys": len(mon.get("keys") or []), "orders": len(mon.get("orders") or []), } def _position_float_pnl(pos: dict) -> float: for key in ("unrealized_pnl", "unrealizedPnl", "upnl"): v = _safe_float(pos.get(key)) if v is not None: return v return 0.0 def _collect_open_issues( *, monitored: bool, agent_ok: bool, flask_ok: bool, positions: list, hub_mon: Optional[dict], day_pnl: float, ) -> list[str]: issues: list[str] = [] if not monitored: return issues if not agent_ok: issues.append("Agent 连接异常") if not flask_ok: issues.append("Flask 监控连接异常") if day_pnl < -0.01: issues.append(f"当日平仓亏损 {day_pnl:.2f}U") open_positions = _filter_open_positions(positions) float_pnl = sum(_position_float_pnl(p) for p in open_positions) if float_pnl < -0.5: issues.append(f"当前浮亏 {float_pnl:.2f}U") if isinstance(hub_mon, dict) and hub_mon.get("ok") is not False: orders = hub_mon.get("orders") or [] trends = hub_mon.get("trends") or [] if open_positions and not orders and not trends: issues.append("交易所有持仓但无本地 active 监控/趋势计划") return issues def previous_trading_day(trading_day: str) -> str: day = (trading_day or "").strip()[:10] if not day: return day dt = datetime.strptime(day, "%Y-%m-%d") return (dt - timedelta(days=1)).strftime("%Y-%m-%d") def _fmt_fund(v: Any) -> str: n = _safe_float(v) if n is None: return "未知" return f"{n:.2f}U" def _format_trade_line(t: dict, *, day_label: str = "") -> str: prefix = f"[{day_label}] " if day_label else "" return ( f"{prefix}{t.get('symbol')} {t.get('direction')} {t.get('result')} " f"{t.get('pnl_amount')}U @ {t.get('closed_at') or '?'}" ) def _monitor_label(item: dict, default: str = "") -> str: for key in ("monitor_type_label", "monitor_type", "entry_reason", "source_label"): val = item.get(key) if val: return str(val) return default def _format_monitor_sections(hub_mon: Optional[dict]) -> dict[str, list[str]]: out = {"trends": [], "orders": [], "keys": [], "rolls": []} if not isinstance(hub_mon, dict) or hub_mon.get("ok") is False: return out for t in hub_mon.get("trends") or []: if not isinstance(t, dict): continue out["trends"].append( f"{t.get('symbol')} {t.get('direction')} " f"SL={t.get('stop_loss')} TP={t.get('take_profit')} " f"补仓区[{t.get('add_lower')}~{t.get('add_upper')}] " f"状态={t.get('status')}" ) for o in hub_mon.get("orders") or []: if not isinstance(o, dict): continue label = _monitor_label(o, "下单监控") out["orders"].append( f"{label}: {o.get('symbol')} {o.get('direction')} " f"触发={o.get('trigger_price')} SL={o.get('stop_loss')} TP={o.get('take_profit')} " f"状态={o.get('status')}" ) for k in hub_mon.get("keys") or []: if not isinstance(k, dict): continue out["keys"].append( f"关键位: {k.get('symbol')} {k.get('direction')} " f"上={k.get('upper')} 下={k.get('lower')} 类型={k.get('monitor_type')}" ) for r in hub_mon.get("rolls") or []: if not isinstance(r, dict): continue out["rolls"].append( f"顺势加仓: {r.get('symbol')} {r.get('direction')} " f"腿数={r.get('leg_count')} SL={r.get('current_stop_loss') or r.get('initial_stop_loss')} " f"状态={r.get('status')}" ) return out def _fetch_account_bundle(client: httpx.Client, ex: dict, trading_day: str) -> dict[str, Any]: name = ex.get("name") or ex.get("key") or ex.get("id") key = ex.get("key") or "" enabled = bool(ex.get("enabled")) env_disabled = bool(ex.get("env_disabled")) monitored = enabled and not env_disabled base: dict[str, Any] = { "id": ex.get("id"), "key": key, "name": name, "enabled": enabled, "env_disabled": env_disabled, "status": "未监控" if not monitored else "已监控", "trades": [], "trade_stats": summarize_trades([]), "positions": [], "open_position_count": 0, "float_pnl_u": 0.0, "balance_usdt": None, "funding_usdt": None, "trading_usdt": None, "available_trading_usdt": None, "trades_yesterday": [], "trade_stats_yesterday": summarize_trades([]), "monitor_lines": {"trends": [], "orders": [], "keys": [], "rolls": []}, "issues": [], "agent_ok": False, "flask_ok": False, "hub_monitor": None, "active_orders": 0, "active_trends": 0, } if not monitored: base["issues"] = [] return base agent_url = (ex.get("agent_url") or "").rstrip("/") flask_url = (ex.get("flask_url") or "").rstrip("/") agent_body = None if agent_url: try: r = client.get( f"{agent_url}/status", headers=_agent_headers(), timeout=hub_agent_timeout(), ) if r.status_code == 200: agent_body = r.json() base["agent_ok"] = True except Exception as exc: base["issues"].append(f"Agent: {exc}") if isinstance(agent_body, dict): base["balance_usdt"] = _safe_float(agent_body.get("balance_usdt")) positions = agent_body.get("positions") or [] if isinstance(positions, list): open_positions = _filter_open_positions(positions) base["positions"] = open_positions base["open_position_count"] = len(open_positions) base["float_pnl_u"] = round(sum(_position_float_pnl(p) for p in open_positions), 4) hub_mon = None prev_day = previous_trading_day(trading_day) if flask_url: try: r = client.get( f"{flask_url}/api/hub/account", headers=_hub_headers(), timeout=hub_flask_timeout(), ) if r.status_code == 200: acct_body = r.json() if isinstance(acct_body, dict) and acct_body.get("ok"): base["funding_usdt"] = _safe_float(acct_body.get("funding_usdt")) base["trading_usdt"] = _safe_float(acct_body.get("trading_usdt")) base["available_trading_usdt"] = _safe_float(acct_body.get("available_trading_usdt")) base["flask_ok"] = True except Exception as exc: base["issues"].append(f"资金接口: {exc}") try: r = client.get( f"{flask_url}/api/hub/trades/today", headers=_hub_headers(), params={"trading_day": trading_day}, timeout=hub_flask_timeout(), ) if r.status_code == 200: trades_body = r.json() if isinstance(trades_body, dict) and trades_body.get("ok"): base["trades"] = trades_body.get("trades") or [] base["trade_stats"] = trades_body.get("stats") or summarize_trades(base["trades"]) base["flask_ok"] = True except Exception as exc: base["issues"].append(f"成交接口: {exc}") if prev_day: try: r = client.get( f"{flask_url}/api/hub/trades/today", headers=_hub_headers(), params={"trading_day": prev_day}, timeout=hub_flask_timeout(), ) if r.status_code == 200: y_body = r.json() if isinstance(y_body, dict) and y_body.get("ok"): base["trades_yesterday"] = y_body.get("trades") or [] base["trade_stats_yesterday"] = y_body.get("stats") or summarize_trades( base["trades_yesterday"] ) base["flask_ok"] = True except Exception as exc: base["issues"].append(f"昨日成交: {exc}") try: r = client.get( f"{flask_url}/api/hub/monitor", headers=_hub_headers(), timeout=hub_flask_timeout(), ) if r.status_code == 200: hub_mon = r.json() if isinstance(hub_mon, dict) and hub_mon.get("ok") is not False: base["hub_monitor"] = hub_mon base["flask_ok"] = True base["active_orders"] = len(hub_mon.get("orders") or []) base["active_trends"] = len(hub_mon.get("trends") or []) base["monitor_lines"] = _format_monitor_sections(hub_mon) except Exception as exc: if "成交接口" not in str(base["issues"]): base["issues"].append(f"监控接口: {exc}") if monitored and not base["agent_ok"] and not base["flask_ok"]: base["status"] = "连接异常" elif base["issues"]: base["status"] = "已监控·需关注" day_pnl = float((base.get("trade_stats") or {}).get("total_pnl_u") or 0) base["issues"].extend( _collect_open_issues( monitored=monitored, agent_ok=base["agent_ok"], flask_ok=base["flask_ok"], positions=base["positions"], hub_mon=hub_mon if isinstance(hub_mon, dict) else None, day_pnl=day_pnl, ) ) base["issues"] = list(dict.fromkeys(base["issues"])) return base def build_daily_context( exchanges: list[dict], *, trading_day: Optional[str] = None, ) -> dict[str, Any]: day = (trading_day or "").strip()[:10] or current_trading_day( reset_hour=trading_day_reset_hour() ) accounts: list[dict] = [] with httpx.Client() as client: for ex in exchanges or []: accounts.append(_fetch_account_bundle(client, ex, day)) total_closed_pnl = 0.0 total_closed = total_win = total_loss = 0 total_float = 0.0 total_funding = 0.0 total_trading = 0.0 total_open_positions = 0 funding_known = trading_known = 0 for ac in accounts: if ac.get("status") == "未监控": continue st = ac.get("trade_stats") or {} total_closed_pnl += float(st.get("total_pnl_u") or 0) total_closed += int(st.get("closed_count") or 0) total_win += int(st.get("win_count") or 0) total_loss += int(st.get("loss_count") or 0) total_float += float(ac.get("float_pnl_u") or 0) total_open_positions += int(ac.get("open_position_count") or _account_open_position_count(ac)) fu = _safe_float(ac.get("funding_usdt")) tu = _safe_float(ac.get("trading_usdt")) if fu is not None: total_funding += fu funding_known += 1 if tu is not None: total_trading += tu trading_known += 1 if not funding_known: total_funding = None if not trading_known: total_trading = None totals = { "trading_day": day, "prev_trading_day": previous_trading_day(day), "total_pnl_u": round(total_closed_pnl, 4), "closed_count": total_closed, "win_count": total_win, "loss_count": total_loss, "float_pnl_u": round(total_float, 4), "open_position_count": total_open_positions, "total_funding_usdt": round(total_funding, 4) if total_funding is not None else None, "total_trading_usdt": round(total_trading, 4) if total_trading is not None else None, } record_fund_snapshot(day, accounts, keep_days=FUND_HISTORY_DAYS) fund_history = get_fund_history(anchor_day=day, keep_days=FUND_HISTORY_DAYS) account_names = {str(ac.get("key") or ac.get("id")): ac.get("name") for ac in accounts} fund_history_text = format_fund_history_text(fund_history, account_names=account_names) payload = { "trading_day": day, "prev_trading_day": previous_trading_day(day), "totals": totals, "accounts": accounts, "fund_history": fund_history, "fund_history_text": fund_history_text, } text = format_context_text(payload) digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] return { "trading_day": day, "prev_trading_day": previous_trading_day(day), "totals": totals, "accounts": accounts, "fund_history": fund_history, "fund_history_text": fund_history_text, "text": text, "context_hash": digest, } def format_context_text(payload: dict) -> str: lines = [] totals = payload.get("totals") or {} day = totals.get("trading_day") prev_day = totals.get("prev_trading_day") or previous_trading_day(str(day or "")) lines.append( f"【合计·今日 {day}】平仓盈亏 {totals.get('total_pnl_u')}U | " f"笔数 {totals.get('closed_count')}（胜{totals.get('win_count')}/负{totals.get('loss_count')}）| " f"实盘持仓 {totals.get('open_position_count', 0)} 仓 | " f"浮盈亏 {totals.get('float_pnl_u')}U | " f"资金账户合计 {_fmt_fund(totals.get('total_funding_usdt'))} | " f"交易账户合计 {_fmt_fund(totals.get('total_trading_usdt'))}" ) lines.append( f"【对比交易日】昨日={prev_day}，今日={day}。" "「持仓」= 交易所 Agent 实盘；「趋势/关键位/监控单/加仓」= 本地计划，不等于已开仓。" ) fund_txt = str(payload.get("fund_history_text") or "").strip() if fund_txt: lines.append("") lines.append(fund_txt) lines.append("") for ac in payload.get("accounts") or []: st = ac.get("trade_stats") or {} sty = ac.get("trade_stats_yesterday") or {} lines.append(f"--- 账户：{ac.get('name')} ({ac.get('key')}) ---") lines.append(f"状态：{ac.get('status')}") if ac.get("status") == "未监控": lines.append("") continue lines.append( f"资金账户 {_fmt_fund(ac.get('funding_usdt'))} | " f"交易账户 {_fmt_fund(ac.get('trading_usdt'))} | " f"可用 {_fmt_fund(ac.get('available_trading_usdt'))}" ) lines.append( f"今日({day})平仓：{st.get('closed_count')} 笔，盈亏 {st.get('total_pnl_u')}U " f"（胜{st.get('win_count')}/负{st.get('loss_count')}）" ) lines.append( f"昨日({prev_day})平仓：{sty.get('closed_count')} 笔，盈亏 {sty.get('total_pnl_u')}U " f"（胜{sty.get('win_count')}/负{sty.get('loss_count')}）" ) open_n = int(ac.get("open_position_count") or _account_open_position_count(ac)) if open_n <= 0: lines.append("当前交易所持仓：无（空仓）") else: lines.append( f"当前交易所持仓：{open_n} 仓 | 浮盈亏合计 {ac.get('float_pnl_u')}U" ) mon = ac.get("monitor_lines") or {} if mon.get("trends"): lines.append("趋势回调计划（本地，非持仓）：") for row in mon["trends"][:8]: lines.append(f" - {row}") if mon.get("rolls"): lines.append("顺势加仓（本地，非持仓）：") for row in mon["rolls"][:8]: lines.append(f" - {row}") if mon.get("keys"): lines.append("关键位监控（本地，非持仓）：") for row in mon["keys"][:8]: lines.append(f" - {row}") if mon.get("orders"): lines.append("进行中的下单监控（本地，非持仓）：") for row in mon["orders"][:8]: lines.append(f" - {row}") positions = ac.get("positions") or [] if positions: lines.append("持仓明细（交易所实盘）：") for p in positions[:8]: if not isinstance(p, dict): continue sym = p.get("symbol") or "?" side = p.get("side") or "?" contracts = p.get("contracts") or p.get("size") or "?" upnl = _position_float_pnl(p) lines.append(f" - {sym} {side} 张数{contracts} 浮盈亏{upnl:.4f}U") lines.append( f"Agent合约余额：{ac.get('balance_usdt') if ac.get('balance_usdt') is not None else '未知'} USDT" ) trades_today = ac.get("trades") or [] if trades_today: lines.append(f"今日平仓明细：") for t in trades_today[:15]: lines.append(f" - {_format_trade_line(t)}") trades_y = ac.get("trades_yesterday") or [] if trades_y: lines.append(f"昨日平仓明细：") for t in trades_y[:15]: lines.append(f" - {_format_trade_line(t)}") if not trades_today and not trades_y: lines.append("平仓明细：无") issues = ac.get("issues") or [] if issues: lines.append("关注点：" + "；".join(issues)) lines.append("") return "\n".join(lines).strip() def format_account_remark(ac: dict) -> str: """分户表格备注：监控摘要 + 持仓。""" parts: list[str] = [] mon = ac.get("monitor_lines") or {} if mon.get("trends"): parts.append(f"趋势{len(mon['trends'])}") if mon.get("rolls"): parts.append(f"加仓{len(mon['rolls'])}") if mon.get("keys"): parts.append(f"关键位{len(mon['keys'])}") if mon.get("orders"): parts.append(f"监控单{len(mon['orders'])}") positions = ac.get("positions") or [] if positions: for p in positions[:2]: if not isinstance(p, dict): continue sym = p.get("symbol") or "?" side = p.get("side") or "?" upnl = _position_float_pnl(p) parts.append(f"{sym} {side} 浮{upnl:.2f}U") if len(positions) > 2: parts.append(f"+{len(positions) - 2}仓") if not parts: issues = ac.get("issues") or [] if issues: return "；".join(str(x) for x in issues[:2]) return "无" return "；".join(parts) def collect_closed_trades_snapshot(accounts: list[dict], *, today: str, yesterday: str) -> list[dict]: rows: list[dict] = [] for ac in accounts or []: name = ac.get("name") or ac.get("key") for t in ac.get("trades_yesterday") or []: if not isinstance(t, dict): continue rows.append({**t, "account_name": name, "trading_day": yesterday}) for t in ac.get("trades") or []: if not isinstance(t, dict): continue rows.append({**t, "account_name": name, "trading_day": today}) rows.sort(key=lambda x: str(x.get("closed_at") or x.get("opened_at") or ""), reverse=True) return rows[:80] def format_chat_position_overview(payload: dict) -> str: totals = payload.get("totals") or {} total_open = int(totals.get("open_position_count") or 0) if total_open <= 0: head = f"【实盘持仓总览】当前空仓（监控户合计 0 仓）。浮盈亏 0U 表示无持仓，不是「有仓但不动」。" else: head = ( f"【实盘持仓总览】监控户合计 {total_open} 仓，" f"浮盈亏合计 {totals.get('float_pnl_u')}U。" ) lines = [ head, "【区分】只有带「持仓明细/交易所实盘」字样的才是已开仓；趋势回调、关键位、下单监控、顺势加仓是本地计划/监控，不算持仓。", ] for ac in payload.get("accounts") or []: if ac.get("status") == "未监控": continue n = int(ac.get("open_position_count") or _account_open_position_count(ac)) mc = _monitor_counts(ac) mon_parts = [] if mc["trends"]: mon_parts.append(f"趋势{mc['trends']}") if mc["rolls"]: mon_parts.append(f"加仓{mc['rolls']}") if mc["keys"]: mon_parts.append(f"关键位{mc['keys']}") if mc["orders"]: mon_parts.append(f"监控单{mc['orders']}") mon_txt = f"；本地监控 {' '.join(mon_parts)}" if mon_parts else "" if n <= 0: lines.append(f"- {ac.get('name')}：空仓{mon_txt}") else: lines.append( f"- {ac.get('name')}：{n}仓 浮盈亏{ac.get('float_pnl_u')}U{mon_txt}" ) return "\n".join(lines) def format_chat_context_for_chat(payload: dict, max_chars: int = 5200) -> str: overview = format_chat_position_overview(payload) body = format_context_text(payload) text = overview + "\n\n" + body if len(text) <= max_chars: return text budget = max(800, max_chars - len(overview) - 4) return overview + "\n\n" + body[:budget].rstrip() + "..." def format_chat_context_brief(payload: dict, max_chars: int = 4500) -> str: return format_chat_context_for_chat(payload, max_chars=max_chars)