网络流量监控模块（一期）Implementation Plan

> For Claude: REQUIRED SUB-SKILL: 使用 executing-plans 或 subagent-driven-development 技能逐任务实现本计划。

Goal: 在网站「系统状态监控」页新增「网络流量监控」板块，回答「今天用手机热点到底烧了多少流量、是哪个模块烧的」。

Architecture: 一期为零侵入的进程级总账。用 macOS 自带 nettop -P -x -l 1 -J bytes_in,bytes_out 采集每个进程的累计收发字节（已实测验证为单调递增计数），两次采样做差分得到窗口流量；用 ps -axo pid,ppid 构建进程树，把 nettop 里的进程沿父链归因到项目模块（解决 Playwright 拉起的系统 Chrome、子进程抓取等「流量大户不在 .pid 里」的问题）；用 netstat -ib 读取默认网卡累计字节做系统总量校验。差分结果按 module 落库到 traffic_history，API 聚合成「实时速率 / 今日构成 / 近30日趋势」。

Tech Stack: Python 3.12、SQLite(quant.db, WAL)、macOS 系统命令 nettop/netstat/ps、原生 http.server、内联 JS（双花括号 {{ }}）。

关键限制（一期，必须在 UI 上标注）:

• unified_server.py（.web_server.pid）同时承载网站 + 游戏交易，一期无法拆分二者流量，合并显示为「🌐 Web服务+游戏交易」。
• 手机浏览器自身的流量无法监控（不在本机进程表内）。
• 仅统计默认网卡（en0）；loopback(lo0) 不计入系统总量。

---

现状锚点（实现前必读）

铁律提醒:

• 修改 unified_server.py 只能用 SearchReplace，禁止 cat/cp 覆盖（沙箱 Operation not permitted）。
• 前端是 .format() 模板：CSS/JS 里的花括号必须写两层 {{ }}。
• 数据管道禁止 except Exception: pass 静默吞异常，必须 logging/print 可观测。

---

Task 1: DB 层 — traffic_history 表与读写函数

Files:

• Modify: adata/aohan/quant/db/system_status.py
• Test: tests/adata_test/utils/__init__.py（新建空文件）、tests/adata_test/utils/test_traffic_monitor.py（新建）

Step 1: 写失败测试

新建 tests/adata_test/utils/__init__.py（空文件，标识包）。

新建 tests/adata_test/utils/test_traffic_monitor.py：

"""网络流量监控 — 解析/归因/差分/落库 单元测试。"""
import os
import sys
import sqlite3
import tempfile
import time
import unittest

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
sys.path.insert(0, PROJECT_ROOT)

import adata.aohan.quant.db.system_status as ss


class TestTrafficDB(unittest.TestCase):
    """traffic_history 表读写与按日聚合。"""

    def setUp(self):
        fd, self.db_path = tempfile.mkstemp(suffix=".db")
        os.close(fd)
        conn = sqlite3.connect(self.db_path)
        conn.executescript(ss.CREATE_SQL)
        conn.commit()
        conn.close()
        self._orig = ss.get_conn

        def _fake_conn():
            c = sqlite3.connect(self.db_path)
            c.row_factory = sqlite3.Row
            return c

        ss.get_conn = _fake_conn

    def tearDown(self):
        ss.get_conn = self._orig
        os.unlink(self.db_path)

    def test_add_and_daily_aggregate(self):
        ss.add_traffic_samples([
            {"module": "fetch", "display_name": "🔄 ETF数据拉取",
             "bytes_in": 1000, "bytes_out": 200, "interval_s": 60.0},
            {"module": "other", "display_name": "🖥️ 系统/其他进程",
             "bytes_in": 500, "bytes_out": 50, "interval_s": 60.0},
        ])
        rows = ss.get_traffic_daily(days=1)
        by_mod = {r["module"]: r for r in rows}
        self.assertEqual(by_mod["fetch"]["in_bytes"], 1000)
        self.assertEqual(by_mod["fetch"]["out_bytes"], 200)
        self.assertEqual(by_mod["other"]["in_bytes"], 500)
        today = time.strftime("%Y-%m-%d")
        self.assertEqual(by_mod["fetch"]["day"], today)

    def test_add_many_rows_then_sum(self):
        rows = [{"module": "fetch", "display_name": "x",
                 "bytes_in": i, "bytes_out": 0, "interval_s": 60.0} for i in range(5)]
        ss.add_traffic_samples(rows)
        agg = {r["module"]: r for r in ss.get_traffic_daily(days=1)}
        self.assertEqual(agg["fetch"]["in_bytes"], 0 + 1 + 2 + 3 + 4)

Step 2: 运行测试确认失败

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v（在项目根目录；若沙箱报 encodings 错误，前面加 env -u PYTHONHOME -u PYTHONPATH） Expected: FAIL — AttributeError: module 'adata.aohan.quant.db.system_status' has no attribute 'add_traffic_samples'

Step 3: 最小实现

在 adata/aohan/quant/db/system_status.py 顶部常量区（TABLE_REQSTATS = 'request_stats_history' 之后）追加：

TABLE_TRAFFIC = 'traffic_history'

在 CREATE_SQL 的 idx_reqstats_ts_route 索引之后（f-string 内部、末尾三引号之前）追加：

    CREATE TABLE IF NOT EXISTS {TABLE_TRAFFIC} (
        id           INTEGER PRIMARY KEY AUTOINCREMENT,
        ts           REAL NOT NULL,
        module       TEXT NOT NULL,
        display_name TEXT,
        bytes_in     INTEGER DEFAULT 0,
        bytes_out    INTEGER DEFAULT 0,
        interval_s   REAL
    );

    CREATE INDEX IF NOT EXISTS idx_traffic_ts ON {TABLE_TRAFFIC}(ts DESC);
    CREATE INDEX IF NOT EXISTS idx_traffic_ts_module ON {TABLE_TRAFFIC}(ts DESC, module);

在文件末尾追加：

# ---------- 网络流量监控 ----------

def add_traffic_samples(rows: List[dict]):
    """批量写入一个采样窗口的各模块流量。rows: [{module, display_name, bytes_in, bytes_out, interval_s}]"""
    now = time.time()
    conn = get_conn()
    try:
        conn.executemany(
            f"INSERT INTO {TABLE_TRAFFIC} (ts, module, display_name, bytes_in, bytes_out, interval_s) "
            f"VALUES (?, ?, ?, ?, ?, ?)",
            [(now, r["module"], r.get("display_name"),
              int(r.get("bytes_in") or 0), int(r.get("bytes_out") or 0),
              r.get("interval_s")) for r in rows]
        )
        conn.execute(
            f"DELETE FROM {TABLE_TRAFFIC} WHERE ts < ?",
            (time.time() - 30 * 86400,)
        )
        conn.commit()
    finally:
        conn.close()


def get_traffic_daily(days: int = 30) -> List[Dict]:
    """按「本地日期 × 模块」聚合收发流量。返回 [{day, module, display_name, in_bytes, out_bytes}]，按日期升序。"""
    conn = get_conn()
    try:
        rows = conn.execute(
            f"SELECT date(ts, 'unixepoch', 'localtime') AS day, module, "
            f"MAX(display_name) AS display_name, "
            f"SUM(COALESCE(bytes_in, 0)) AS in_bytes, "
            f"SUM(COALESCE(bytes_out, 0)) AS out_bytes "
            f"FROM {TABLE_TRAFFIC} WHERE ts >= ? "
            f"GROUP BY day, module ORDER BY day",
            (time.time() - days * 86400,)
        ).fetchall()
        return [dict(r) for r in rows]
    finally:
        conn.close()

Step 4: 运行测试确认通过

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: PASS（2 passed）

Step 5: 提交

git add adata/aohan/quant/db/system_status.py tests/adata_test/utils/__init__.py tests/adata_test/utils/test_traffic_monitor.py
git commit -m "feat(traffic): traffic_history 表 + 按日聚合读写函数"

---

Task 2: 采样器纯函数 — 解析 / 归因 / 差分

Files:

• Create: adata/aohan/quant/utils/traffic_monitor.py
• Test: tests/adata_test/utils/test_traffic_monitor.py（追加用例）

Step 1: 写失败测试

在 tests/adata_test/utils/test_traffic_monitor.py 末尾追加：

from adata.aohan.quant.utils import traffic_monitor as tm


NETTOP_SAMPLE = """\
                                                                                bytes_in       bytes_out
launchd.1                                                                              0               0
Python.4199                                                                     10485760         2097152
Python.21562                                                                  165562936005     63448417867
Google Chrome.9999                                                               52428800         1048576
"""

NETSTAT_SAMPLE = """\
Name       Mtu   Network       Address            Ipkts Ierrs     Ibytes    Opkts Oerrs     Obytes  Coll
lo0        16384 <Link#1>                      24271751     0 73297935873 24271751     0 73297935873     0
en0        1500  <Link#6>    14:7d:da:d4:7e:93 221324263     0 165562936005 195562666     0 63448417867     0
"""

PS_TREE_SAMPLE = """\
  PID  PPID
    1     0
21545     1
21562 21545
 9999 21562
 4199 21545
"""


class TestParsers(unittest.TestCase):
    def test_parse_nettop(self):
        d = tm.parse_nettop_output(NETTOP_SAMPLE)
        self.assertEqual(d[4199], (10485760, 2097152))
        self.assertEqual(d[21562], (165562936005, 63448417867))
        self.assertEqual(d[9999], (52428800, 1048576))
        self.assertNotIn(1, d)  # launchd 全 0 也在，但 header 行必须被跳过

    def test_parse_netstat_en0_with_mac(self):
        self.assertEqual(tm.parse_netstat_link(NETSTAT_SAMPLE, "en0"),
                         (165562936005, 63448417867))

    def test_parse_netstat_lo0_without_mac(self):
        """lo0 行没有 Address 列，列位会左移 —— 必须按 <Link#> 定位而非固定下标。"""
        self.assertEqual(tm.parse_netstat_link(NETSTAT_SAMPLE, "lo0"),
                         (73297935873, 73297935873))

    def test_parse_netstat_missing_iface(self):
        self.assertIsNone(tm.parse_netstat_link(NETSTAT_SAMPLE, "utun3"))

    def test_parse_ps_tree(self):
        tree = tm.parse_ps_tree(PS_TREE_SAMPLE)
        self.assertEqual(tree[21562], 21545)
        self.assertEqual(tree[9999], 21562)

    def test_attribute_self(self):
        tree = tm.parse_ps_tree(PS_TREE_SAMPLE)
        self.assertEqual(tm.attribute_pid(4199, tree, {4199: "fetch"}), "fetch")

    def test_attribute_via_parent_chain(self):
        """Chrome(9999) 的父链 9999→21562 命中 web_daemon。"""
        tree = tm.parse_ps_tree(PS_TREE_SAMPLE)
        self.assertEqual(tm.attribute_pid(9999, tree, {21562: "web_daemon"}), "web_daemon")

    def test_attribute_unknown(self):
        tree = tm.parse_ps_tree(PS_TREE_SAMPLE)
        self.assertEqual(tm.attribute_pid(9999, tree, {4199: "fetch"}), tm.OTHER_MODULE)

    def test_attribute_no_infinite_loop(self):
        self.assertEqual(tm.attribute_pid(5, {5: 5}, {}), tm.OTHER_MODULE)

    def test_diff_counters(self):
        prev = {1: (100, 200), 2: (50, 0)}
        cur = {1: (160, 260), 3: (10, 10)}
        d = tm.diff_counters(cur, prev)
        self.assertEqual(d[1], (60, 60))
        self.assertEqual(d[3], (10, 10))       # 新进程等值差分 → 按当前值
        self.assertNotIn(2, d)                  # 消失的进程不输出

    def test_diff_counters_reset(self):
        """进程重启导致计数回退 → 按当前值计（不产生负流量）。"""
        d = tm.diff_counters({1: (30, 40)}, {1: (100, 200)})
        self.assertEqual(d[1], (30, 40))

    def test_read_module_pids(self):
        with tempfile.TemporaryDirectory() as d:
            with open(os.path.join(d, ".run_fetch.pid"), "w") as f:
                f.write("4199\n")
            with open(os.path.join(d, ".web_server.pid"), "w") as f:
                f.write("21562\n")
            with open(os.path.join(d, ".run_pushd.pid"), "w") as f:
                f.write("not-a-number\n")   # 脏数据必须被跳过
            m = tm.read_module_pids(d)
            self.assertEqual(m[4199], "fetch")
            self.assertEqual(m[21562], "web_daemon")
            self.assertEqual(len(m), 2)

Step 2: 运行测试确认失败

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: FAIL — ModuleNotFoundError: No module named 'adata.aohan.quant.utils.traffic_monitor'

Step 3: 最小实现

新建 adata/aohan/quant/utils/traffic_monitor.py：

#!/usr/bin/env python3
"""网络流量监控 — 基于 nettop 进程级累计计数差分，按模块归因，落库 + 查询。

原理：
  `nettop -P -x -l 1 -J bytes_in,bytes_out` 输出每个进程的「累计」收发字节（实测单调递增，
  已用 --limit-rate 300K 的 curl 验证）。两次采样差分 → 窗口内该进程流量；
  再用 `ps -axo pid,ppid` 构建进程树，把子进程（Playwright 拉起的系统 Chrome 等）
  沿父链归因到项目模块。`netstat -ib` 提供网卡级累计字节用于系统总量校验。

一期限制：unified_server.py 同时承载「网站 + 游戏交易」，二者流量合并统计。
"""
import os
import re
import time
import threading
import subprocess
from typing import Dict, List, Optional, Tuple

_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))

# PID 文件 → (module_key, display_name)
PID_FILE_MAP = [
    (".run_fetch.pid", "fetch", "🔄 ETF数据拉取"),
    (".run_stock_fetchd.pid", "stock_fetchd", "📊 股票选股"),
    (".run_pushd.pid", "pushd", "📨 推送调度"),
    (".run_market_observe.pid", "market_observe", "📈 行情观察"),
    (".web_server.pid", "web_daemon", "🌐 Web服务+游戏交易"),
    (".web_tunnel.pid", "tunnel", "🔒 Cloudflare隧道"),
    (".tushare_probe.pid", "tushare_probe", "🔍 Tushare探测"),
]

MODULE_DISPLAY = {key: disp for _f, key, disp in PID_FILE_MAP}
OTHER_MODULE = "other"
OTHER_DISPLAY = "🖥️ 系统/其他进程"

_MAC_RE = re.compile(r'^([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$')


def _module_display(module: str) -> str:
    return MODULE_DISPLAY.get(module, OTHER_DISPLAY)


# ---------------- 纯解析函数（可单测） ----------------

def parse_nettop_output(text: str) -> Dict[int, Tuple[int, int]]:
    """解析 nettop 输出 → {pid: (bytes_in, bytes_out)}。"""
    result: Dict[int, Tuple[int, int]] = {}
    for line in (text or "").splitlines():
        parts = line.split()
        if len(parts) < 3:
            continue
        pid_str = parts[0].rsplit(".", 1)[-1]
        if not pid_str.isdigit():
            continue
        try:
            result[int(pid_str)] = (int(parts[-2]), int(parts[-1]))
        except (ValueError, IndexError):
            continue
    return result


def parse_netstat_link(text: str, iface: str) -> Optional[Tuple[int, int]]:
    """解析 netstat -ib 输出中指定网卡的 <Link#> 合计行 → (ibytes, obytes)。

    注意：en0 行有 Address(MAC) 列、lo0 行没有 —— 必须用 <Link#> 定位，再按列位取数，
    不能假定固定下标。
    """
    for line in (text or "").splitlines():
        parts = line.split()
        if len(parts) < 4 or parts[0] != iface:
            continue
        li = next((i for i, p in enumerate(parts) if p.startswith("<Link#")), None)
        if li is None:
            continue
        rest = parts[li + 1:]
        if rest and _MAC_RE.match(rest[0]):
            rest = rest[1:]
        if len(rest) < 6:
            continue
        try:
            return int(rest[2]), int(rest[5])   # Ibytes, Obytes
        except (ValueError, IndexError):
            continue
    return None


def parse_ps_tree(text: str) -> Dict[int, int]:
    """解析 `ps -axo pid,ppid` 输出 → {pid: ppid}。"""
    tree: Dict[int, int] = {}
    for line in (text or "").splitlines()[1:]:
        parts = line.split()
        if len(parts) < 2:
            continue
        try:
            tree[int(parts[0])] = int(parts[1])
        except ValueError:
            continue
    return tree


def attribute_pid(pid: int, pid_tree: Dict[int, int], pid_to_module: Dict[int, str]) -> str:
    """PID 归因到模块：先看自身，再沿父链上溯；找不到 → OTHER_MODULE。防环。"""
    seen = set()
    cur = pid
    while cur and cur not in seen:
        if cur in pid_to_module:
            return pid_to_module[cur]
        seen.add(cur)
        cur = pid_tree.get(cur)
    return OTHER_MODULE


def diff_counters(cur: Dict, prev: Dict) -> Dict:
    """{key: (in,out)} 累计计数差分。计数回退（进程重启）按当前值计；消失的 key 不输出。"""
    out = {}
    for k, (ci, co) in cur.items():
        pi, po = prev.get(k, (ci, co))
        di, do = ci - pi, co - po
        out[k] = (di if di >= 0 else ci, do if do >= 0 else co)
    return out


# ---------------- 命令执行 ----------------

def _bin(path: str, name: str) -> str:
    return path if os.path.exists(path) else name


_NETTOP = _bin("/usr/bin/nettop", "nettop")
_NETSTAT = _bin("/usr/sbin/netstat", "netstat")
_PS = _bin("/bin/ps", "ps")


def _run(cmd: List[str], timeout: float = 8.0) -> str:
    try:
        env = os.environ.copy()
        env.pop("PYTHONHOME", None)
        env.pop("PYTHONPATH", None)
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env)
        return r.stdout or ""
    except Exception as e:
        print(f"[traffic_monitor] run {cmd} error: {e}")
        return ""


def get_default_iface() -> str:
    """从路由表解析默认出口网卡（如 en0）。"""
    out = _run([_NETSTAT, "-rn"])
    for line in out.splitlines():
        parts = line.split()
        if parts and parts[0] == "default":
            for p in reversed(parts):
                if re.match(r'^[a-z]+\d+$', p):
                    return p
    return "en0"


def read_module_pids(base_dir: str = _PROJECT_ROOT) -> Dict[int, str]:
    """读取各守护进程 .pid 文件 → {pid: module_key}。脏数据跳过。"""
    mapping: Dict[int, str] = {}
    for fname, key, _disp in PID_FILE_MAP:
        path = os.path.join(base_dir, fname)
        try:
            with open(path) as f:
                pid = int(f.read().strip().split()[0])
            if pid > 0:
                mapping[pid] = key
        except (OSError, ValueError, IndexError):
            continue
    return mapping

Step 4: 运行测试确认通过

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: PASS（14 passed）

Step 5: 提交

git add adata/aohan/quant/utils/traffic_monitor.py tests/adata_test/utils/test_traffic_monitor.py
git commit -m "feat(traffic): nettop/netstat/ps 解析 + 进程树归因 + 计数差分"

---

Task 3: 采样循环 + 快照聚合

Files:

• Modify: adata/aohan/quant/utils/traffic_monitor.py（追加采样/聚合）
• Test: tests/adata_test/utils/test_traffic_monitor.py（追加用例）

Step 1: 写失败测试

追加：

class TestAggregate(unittest.TestCase):
    """sample_once 的窗口聚合：给定两次计数，验证按模块归属与差分。"""

    def test_sample_once_attributes_children_and_persists(self):
        tm._LAST_SAMPLE = {"ts": 0.0, "nettop": {}, "netstat": None}
        tm._LATEST_SNAPSHOT = {"ts": 0.0, "data": None}
        persisted = []
        tm._persist = lambda rows, ts: persisted.append((rows, ts))

        nettop_out = ("h\nPython.21562 1000 2000\n"
                      "Google Chrome.9999 500 100\n"
                      "Python.4199 300 40\n")
        netstat_out = NETSTAT_SAMPLE
        ps_out = PS_TREE_SAMPLE

        tm._run = lambda cmd, timeout=8.0: (
            nettop_out if cmd and cmd[0] == tm._NETTOP else
            netstat_out if cmd and cmd[0] == tm._NETSTAT else ps_out)
        tm.read_module_pids = lambda base_dir=tm._PROJECT_ROOT: {21562: "web_daemon", 4199: "fetch"}

        # 第一次采样：建立基线，不外泄流量
        s1 = tm.sample_once()
        self.assertTrue(s1["first_sample"])
        self.assertEqual(len(persisted), 0)

        # 第二次采样：Python.21562 +1000/+2000，Chrome 子进程归入 web_daemon
        nettop_out2 = ("h\nPython.21562 2000 3800\n"
                       "Google Chrome.9999 1300 180\n"
                       "Python.4199 500 90\n")
        tm._run = lambda cmd, timeout=8.0: (
            nettop_out2 if cmd and cmd[0] == tm._NETTOP else
            netstat_out if cmd and cmd[0] == tm._NETSTAT else ps_out)
        s2 = tm.sample_once()
        self.assertFalse(s2["first_sample"])
        mods = {m["module"]: m for m in s2["modules_window"]}
        self.assertEqual(mods["web_daemon"]["in_bytes"], 1000 + 800)   # 自身 + Chrome 子进程
        self.assertEqual(mods["fetch"]["in_bytes"], 200)
        self.assertEqual(len(persisted), 1)

    def test_get_traffic_metrics_shape(self):
        tm._LATEST_SNAPSHOT = {"ts": time.time(), "data": {
            "collected_at": time.time(), "interval_s": 60.0, "iface": "en0",
            "first_sample": False,
            "system_window": {"in_bytes": 6000, "out_bytes": 3000},
            "modules_window": [], "top_processes": [],
        }}
        tm._LAST_SAMPLE = {"ts": time.time(), "nettop": {}, "netstat": (100000, 50000)}
        tm._db_daily = lambda days=30: [
            {"day": time.strftime("%Y-%m-%d"), "module": "web_daemon",
             "display_name": "🌐 Web服务+游戏交易", "in_bytes": 10000, "out_bytes": 4000},
            {"day": time.strftime("%Y-%m-%d"), "module": "fetch",
             "display_name": "🔄 ETF数据拉取", "in_bytes": 2000, "out_bytes": 500},
        ]
        m = tm.get_traffic_metrics()
        self.assertEqual(m["today"]["total_bytes"], 16500)
        self.assertAlmostEqual(m["realtime"]["in_bps"], 100.0, places=3)
        self.assertEqual(m["today"]["modules"][0]["module"], "web_daemon")
        self.assertIn("daily", m)
        self.assertIn("iface_total", m)

Step 2: 运行测试确认失败

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: FAIL — AttributeError: module 'adata.aohan.quant.utils.traffic_monitor' has no attribute 'sample_once'

Step 3: 最小实现

在 traffic_monitor.py 末尾追加：

# ---------------- 采样循环 ----------------

_SAMPLE_LOCK = threading.Lock()
_LAST_SAMPLE = {"ts": 0.0, "nettop": {}, "netstat": None}
_LATEST_SNAPSHOT = {"ts": 0.0, "data": None}

_GIB = 1024 ** 3


def _persist(rows: List[dict], ts: float):
    from adata.aohan.quant.db.system_status import add_traffic_samples, init_system_status_tables
    init_system_status_tables()
    add_traffic_samples(rows)


def _db_daily(days: int = 30) -> List[dict]:
    from adata.aohan.quant.db.system_status import get_traffic_daily, init_system_status_tables
    init_system_status_tables()
    return get_traffic_daily(days)


def sample_once() -> dict:
    """采集一次，与上次采样差分并落库，返回本次快照。"""
    now = time.time()
    procs = parse_nettop_output(_run([_NETTOP, "-P", "-x", "-l", "1", "-J", "bytes_in,bytes_out"]))
    iface = get_default_iface()
    iface_counters = parse_netstat_link(_run([_NETSTAT, "-ib"]), iface)
    tree = parse_ps_tree(_run([_PS, "-axo", "pid,ppid"]))

    with _SAMPLE_LOCK:
        prev = _LAST_SAMPLE
        first = prev["ts"] <= 0.0
        interval = (now - prev["ts"]) if not first else None
        proc_delta = {} if first else diff_counters(procs, prev["nettop"])
        iface_delta = None
        if not first and iface_counters and prev["netstat"]:
            iface_delta = diff_counters({iface: iface_counters},
                                        {iface: prev["netstat"]}).get(iface)
        _LAST_SAMPLE["ts"] = now
        _LAST_SAMPLE["nettop"] = procs
        _LAST_SAMPLE["netstat"] = iface_counters
        iface_total = iface_counters

    module_pids = read_module_pids()
    per_module: Dict[str, List[int]] = {}
    per_proc = []
    for pid, (din, dout) in proc_delta.items():
        mod = attribute_pid(pid, tree, module_pids)
        agg = per_module.setdefault(mod, [0, 0])
        agg[0] += din
        agg[1] += dout
        if din or dout:
            per_proc.append({"pid": pid, "module": mod,
                             "module_name": _module_display(mod),
                             "in_bytes": din, "out_bytes": dout})

    if interval and interval > 0 and per_module:
        rows = [{"module": m, "display_name": _module_display(m),
                 "bytes_in": v[0], "bytes_out": v[1], "interval_s": interval}
                for m, v in per_module.items()]
        try:
            _persist(rows, now)
        except Exception as e:
            print(f"[traffic_monitor] persist error: {e}")

    modules_window = [
        {"module": m, "display_name": _module_display(m),
         "in_bytes": v[0], "out_bytes": v[1], "total_bytes": v[0] + v[1]}
        for m, v in sorted(per_module.items(), key=lambda kv: -(kv[1][0] + kv[1][1]))
    ]

    snapshot = {
        "collected_at": now,
        "interval_s": interval,
        "iface": iface,
        "first_sample": first,
        "system_window": ({"in_bytes": iface_delta[0], "out_bytes": iface_delta[1],
                           "total_bytes": iface_delta[0] + iface_delta[1]}
                          if iface_delta else None),
        "modules_window": modules_window,
        "top_processes": sorted(per_proc, key=lambda p: -(p["in_bytes"] + p["out_bytes"]))[:12],
    }
    with _SAMPLE_LOCK:
        _LATEST_SNAPSHOT["ts"] = now
        _LATEST_SNAPSHOT["data"] = snapshot
    return snapshot


def run_traffic_cycle() -> Optional[dict]:
    """调度器调用入口；异常不抛出以免拖垮主循环。"""
    try:
        return sample_once()
    except Exception as e:
        print(f"[traffic_monitor] cycle error: {e}")
        return None


def get_traffic_metrics(force: bool = False) -> dict:
    """组装 API 快照：实时速率 + 今日构成 + 近30日趋势 + 网卡累计。"""
    with _SAMPLE_LOCK:
        snap = _LATEST_SNAPSHOT["data"]

    if snap is None:
        return {
            "ok": True, "collected_at": time.time(), "interval_s": None,
            "iface": get_default_iface(), "first_sample": True,
            "realtime": {"in_bps": 0, "out_bps": 0},
            "today": {"in_bytes": 0, "out_bytes": 0, "total_bytes": 0, "modules": []},
            "daily": [], "top_processes": [], "modules_window": [],
            "iface_total": None, "level": "normal",
        }

    interval = snap.get("interval_s") or 0
    sys_win = snap.get("system_window") or {"in_bytes": 0, "out_bytes": 0}
    realtime = {
        "in_bps": (sys_win["in_bytes"] / interval) if interval else 0,
        "out_bps": (sys_win["out_bytes"] / interval) if interval else 0,
    }

    daily_rows = _db_daily(30)
    today_key = time.strftime("%Y-%m-%d")

    per_day: Dict[str, dict] = {}
    today_agg: Dict[str, dict] = {}
    for r in daily_rows:
        day = r["day"]
        din, dout = int(r.get("in_bytes") or 0), int(r.get("out_bytes") or 0)
        d = per_day.setdefault(day, {"day": day, "in_bytes": 0, "out_bytes": 0, "total_bytes": 0})
        d["in_bytes"] += din
        d["out_bytes"] += dout
        d["total_bytes"] = d["in_bytes"] + d["out_bytes"]
        if day == today_key:
            m = today_agg.setdefault(r["module"], {
                "module": r["module"],
                "display_name": r.get("display_name") or _module_display(r["module"]),
                "in_bytes": 0, "out_bytes": 0})
            m["in_bytes"] += din
            m["out_bytes"] += dout
            m["total_bytes"] = m["in_bytes"] + m["out_bytes"]

    today_modules = sorted(today_agg.values(), key=lambda x: -x["total_bytes"])
    today_total = sum(m["total_bytes"] for m in today_modules)

    if today_total >= 5 * _GIB:
        level = "critical"
    elif today_total >= 2 * _GIB:
        level = "high"
    elif today_total >= 500 * 1024 * 1024:
        level = "moderate"
    else:
        level = "normal"

    iface_total = None
    with _SAMPLE_LOCK:
        ifc = _LAST_SAMPLE.get("netstat")
    if ifc:
        iface_total = {"in_bytes": ifc[0], "out_bytes": ifc[1], "total_bytes": ifc[0] + ifc[1]}

    return {
        "ok": True,
        "collected_at": snap.get("collected_at"),
        "interval_s": snap.get("interval_s"),
        "iface": snap.get("iface"),
        "first_sample": snap.get("first_sample"),
        "realtime": realtime,
        "today": {"in_bytes": sum(m["in_bytes"] for m in today_modules),
                  "out_bytes": sum(m["out_bytes"] for m in today_modules),
                  "total_bytes": today_total, "modules": today_modules},
        "daily": [per_day[k] for k in sorted(per_day)],
        "modules_window": snap.get("modules_window", []),
        "top_processes": snap.get("top_processes", []),
        "iface_total": iface_total,
        "level": level,
    }


if __name__ == "__main__":
    snap = sample_once()
    print(snap)
    time.sleep(3)
    print(get_traffic_metrics())

> 说明：_persist / _db_daily 是模块级函数，测试会 monkeypatch 掉它们，避免写真实 DB。_run / read_module_pids 同理被 patch。

Step 4: 运行测试确认通过

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: PASS（16 passed）

Step 5: 真机冒烟（可选但推荐）

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -m adata.aohan.quant.utils.traffic_monitor Expected: 打印第一、第二个快照；第二快照 interval_s≈3.0，modules_window 含 web_daemon（自身）等条目，top_processes 非空。

Step 6: 提交

git add adata/aohan/quant/utils/traffic_monitor.py tests/adata_test/utils/test_traffic_monitor.py
git commit -m "feat(traffic): 采样循环 + 快照聚合(get_traffic_metrics)"

---

Task 4: 接入调度器（system_monitor）

Files:

• Modify: adata/aohan/quant/utils/system_monitor.py:356-384
• Test: tests/adata_test/utils/test_traffic_monitor.py（追加轻量用例）

Step 1: 写失败测试

追加（验证调度器留有流量钩子的调用契约）：

import inspect
from adata.aohan.quant.utils import system_monitor as sm


class TestSchedulerHook(unittest.TestCase):
    def test_traffic_cycle_referenced_in_scheduler(self):
        """调度循环内必须调用 traffic_monitor.run_traffic_cycle（否则采样永不发生）。"""
        src = inspect.getsource(sm._scheduler_loop)
        self.assertIn("run_traffic_cycle", src)

    def test_run_traffic_cycle_importable(self):
        from adata.aohan.quant.utils.traffic_monitor import run_traffic_cycle
        self.assertTrue(callable(run_traffic_cycle))

Step 2: 运行测试确认失败

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: FAIL — AssertionError: 'run_traffic_cycle' not found in ...

Step 3: 最小实现

在 system_monitor.py 的 _scheduler_loop() 中，check_all_scripts() 之后、perf 段之前，插入流量采样块。将整个 _scheduler_loop 替换为：

def _scheduler_loop():
    global _scheduler_running
    _scheduler_running = True

    try:
        check_all_scripts()
    except Exception as e:
        print(f"[system_monitor] initial check error: {e}")

    _last_perf_cycle = 0.0
    _PERF_INTERVAL = 300.0
    _last_traffic_ts = 0.0
    _TRAFFIC_INTERVAL = 60.0

    while _scheduler_running:
        try:
            check_all_scripts()
        except Exception as e:
            print(f"[system_monitor] scheduler error: {e}")

        # 网络流量监控：每 60s 采样一次（nettop 累计计数差分），失败不阻塞主循环
        now = time.time()
        if now - _last_traffic_ts >= _TRAFFIC_INTERVAL:
            try:
                from adata.aohan.quant.utils.traffic_monitor import run_traffic_cycle
                run_traffic_cycle()
                _last_traffic_ts = time.time()
            except Exception as e:
                print(f"[system_monitor] traffic cycle error: {e}")

        # 网站性能监控：每 5 分钟跑一次（延迟探针 + 请求耗时窗口落库），失败不阻塞主循环
        now = time.time()
        if now - _last_perf_cycle >= _PERF_INTERVAL:
            try:
                from adata.aohan.quant.utils.web_perf import run_perf_cycle
                run_perf_cycle()
                _last_perf_cycle = time.time()
            except Exception as e:
                print(f"[system_monitor] perf cycle error: {e}")

        time.sleep(30)

Step 4: 运行测试确认通过

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: PASS（18 passed）

Step 5: 提交

git add adata/aohan/quant/utils/system_monitor.py tests/adata_test/utils/test_traffic_monitor.py
git commit -m "feat(traffic): system_monitor 调度循环挂载 60s 流量采样"

---

Task 5: 接入 unified_server.py — 导入 + API 端点

Files:

• Modify: unified_server.py:16-25（导入）、unified_server.py:14671（API 分支）
• Test: tests/adata_test/utils/test_traffic_monitor.py（追加源码契约用例）

Step 1: 写失败测试

追加：

class TestServerWiring(unittest.TestCase):
    def _src(self):
        with open(os.path.join(PROJECT_ROOT, "unified_server.py"), encoding="utf-8") as f:
            return f.read()

    def test_imports_get_traffic_metrics(self):
        self.assertIn("get_traffic_metrics as _get_traffic_metrics", self._src())

    def test_api_route_registered(self):
        self.assertIn('path == "/api/system/traffic"', self._src())

    def test_frontend_card_and_loader(self):
        src = self._src()
        self.assertIn('id="sysTraffic"', src)
        self.assertIn("function loadTraffic(", src)
        self.assertIn("renderTraffic(j.traffic)", src)

Step 2: 运行测试确认失败

Run: .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v Expected: FAIL（3 failures — 源码中尚无这些字符串）

Step 3a: 修改导入块

在 unified_server.py 的 system_monitor 导入块里，start_scheduler as _start_system_monitor, 之后追加一行：

    start_scheduler as _start_system_monitor,
    get_traffic_metrics as _get_traffic_metrics,
)

即：

from adata.aohan.quant.utils.system_monitor import (
    get_overview as _get_system_overview,
    get_all_status as _get_system_all_status,
    get_script_history as _get_system_script_history,
    get_all_history as _get_system_all_history,
    get_log_tail as _get_system_log_tail,
    get_host_metrics as _get_host_metrics,
    get_disk_metrics as _get_disk_metrics,
    start_scheduler as _start_system_monitor,
    get_traffic_metrics as _get_traffic_metrics,
)

> 说明：get_traffic_metrics 定义在 traffic_monitor，但它由 system_monitor 统一对外暴露更贴合现有分层。为最小改动，直接在 system_monitor.py 末尾追加一行再导出： > ``python > from adata.aohan.quant.utils.traffic_monitor import get_traffic_metrics # noqa: E402 > ` > 放在 system_monitor.py 文件末尾（if __name__ == "__main__": 之前）。这样 unified_server 从 system_monitor 导入即可，风格一致。

在 unified_server.py 的 /api/system/perf 分支 return 之后（elif path == "/~docs": 之前，约 L14682），插入：

        elif path == "/api/system/traffic":
            try:
                traffic = _get_traffic_metrics()
                body = json.dumps({"ok": True, "traffic": traffic}, ensure_ascii=False, default=str).encode("utf-8")
                self.send_response(200); self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
                return
            except Exception as e:
                self.send_response(500); self.send_header("Content-Type", "application/json; charset=utf-8")
                self.end_headers(); self.wfile.write(json.dumps({"error": str(e)}, ensure_ascii=False).encode("utf-8"))
                return

Run: env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -m py_compile unified_server.py && echo OK Expected: OK

git add unified_server.py adata/aohan/quant/utils/system_monitor.py tests/adata_test/utils/test_traffic_monitor.py
git commit -m "feat(traffic): /api/system/traffic 端点 + 导入接线"

Task 6: /system 页面 — 流量监控卡片 + JS 渲染

• Modify: unified_server.py:6696-6697（卡片 HTML）、unified_server.py:6804-6813（JS 函数区）、unified_server.py:6783-6784（loadStatus 内调用）、unified_server.py:7388-7392（init/定时器/导出）

> 全部改动在 .format() 模板内，新增 CSS/JS 花括号一律写 {{ }}。

在「网站速度监控」卡片结束（</div></div> 结束 sysPerf）之后、sys-etf-stats 之前插入：

  <!-- 网络流量监控 -->
  <div class="sys-traffic" id="sysTraffic" style="margin-bottom:20px;">
    <div class="sys-card"><div class="sys-card-head">
      <div class="sys-card-name">
        <div class="sys-card-icon">📶</div>
        <div><div class="sys-card-title">网络流量监控</div>
          <div style="font-size:11px;color:var(--muted);margin-top:2px;" id="trafficSubtitle">加载中…</div>
        </div>
      </div>
      <div id="trafficBadge" class="sys-status-badge sys-status-unknown">--</div>
    </div>
    <div class="sys-card-body" style="padding:0 18px 16px;">
      <div id="trafficBody" style="font-size:12px;color:var(--muted);line-height:1.8;">加载中…</div>
    </div></div>
  </div>

Step 2: JS — 在 loadPerf 定义之后插入（约 L6814）

  // ===== 网络流量监控 =====
  async function loadTraffic() {{
    try {{
      var r = await fetch('/api/system/traffic');
      var j = await r.json();
      if (!j.ok) {{ document.getElementById('trafficBody').textContent = '加载失败: '+(j.error||''); return; }}
      renderTraffic(j.traffic);
    }} catch(e) {{
      document.getElementById('trafficBody').textContent = '流量数据获取失败: '+e.message;
    }}
  }}

  function fmtBytes(n) {{
    n = Number(n)||0;
    if (n < 1024) return n.toFixed(0)+'B';
    if (n < 1048576) return (n/1024).toFixed(1)+'KB';
    if (n < 1073741824) return (n/1048576).toFixed(1)+'MB';
    return (n/1073741824).toFixed(2)+'GB';
  }}

  function fmtRate(bps) {{
    bps = Number(bps)||0;
    if (bps < 1024) return bps.toFixed(0)+'B/s';
    if (bps < 1048576) return (bps/1024).toFixed(1)+'KB/s';
    return (bps/1048576).toFixed(2)+'MB/s';
  }}

  function dailyBars(daily) {{
    if (!daily || daily.length < 2) return '<div style="margin-top:10px;color:var(--muted);font-size:11px;">暂无历史趋势（每天累计一次）</div>';
    var maxV = Math.max.apply(null, daily.map(function(d){{return d.total_bytes||0;}})) || 1;
    var html = '<div style="margin:12px 0 4px;"><b style="color:var(--text);">📈 近'+daily.length+'日流量</b></div>';
    html += '<div style="max-height:150px;overflow-y:auto;">';
    daily.slice(-14).forEach(function(d) {{
      var w = ((d.total_bytes||0)/maxV*100).toFixed(1);
      html += '<div style="display:flex;align-items:center;gap:8px;margin:2px 0;font-size:11px;">'+
        '<span style="color:var(--muted);width:76px;flex:none;">'+escHtml(d.day)+'</span>'+
        '<span style="flex:1;background:var(--bg2);border-radius:4px;height:10px;overflow:hidden;">'+
          '<span style="display:block;width:'+w+'%;height:100%;background:var(--orange);"></span></span>'+
        '<span style="width:66px;text-align:right;font-variant-numeric:tabular-nums;">'+fmtBytes(d.total_bytes||0)+'</span>'+
      '</div>';
    }});
    html += '</div>';
    return html;
  }}

  function renderTraffic(t) {{
    if (!t) {{ document.getElementById('trafficBody').textContent = '暂无流量数据'; return; }}
    var sub = document.getElementById('trafficSubtitle');
    var badge = document.getElementById('trafficBadge');
    var lvl = t.level||'normal';
    badge.className = 'sys-status-badge sys-status-'+(lvl==='critical'?'stopped':lvl==='high'?'degraded':'running');
    var today = t.today||{{}};
    badge.textContent = fmtBytes(today.total_bytes||0);
    sub.textContent = '今日 ↓'+fmtBytes(today.in_bytes||0)+' / ↑'+fmtBytes(today.out_bytes||0)+
      (t.first_sample ? ' · 等待首次采样…' :
        ' · 实时 ↓'+fmtRate(t.realtime&&t.realtime.in_bps)+' ↑'+fmtRate(t.realtime&&t.realtime.out_bps));

    var html = '';
    html += '<div style="display:flex;gap:24px;flex-wrap:wrap;margin-bottom:10px;">';
    html += '<div><span style="color:var(--muted);">实时下行</span> <b style="font-size:16px;">'+fmtRate(t.realtime&&t.realtime.in_bps)+'</b></div>';
    html += '<div><span style="color:var(--muted);">实时上行</span> <b style="font-size:16px;">'+fmtRate(t.realtime&&t.realtime.out_bps)+'</b></div>';
    html += '<div><span style="color:var(--muted);">今日合计</span> <b style="color:var(--orange);font-size:16px;">'+fmtBytes(today.total_bytes||0)+'</b></div>';
    html += '<div><span style="color:var(--muted);">网卡('+escHtml(t.iface||'en0')+')累计</span> <b style="font-size:16px;">'+fmtBytes(t.iface_total&&t.iface_total.total_bytes)+'</b></div>';
    html += '</div>';

    var mods = (today.modules)||[];
    if (mods.length) {{
      html += '<div style="margin:8px 0 4px;"><b style="color:var(--text);">📊 今日流量构成（按模块）</b></div>';
      html += '<div style="display:grid;grid-template-columns:1fr auto;gap:3px 12px;">';
      mods.forEach(function(m) {{
        var pct = today.total_bytes>0 ? (m.total_bytes/today.total_bytes*100).toFixed(1) : 0;
        html += '<span style="color:var(--text2);">'+escHtml(m.display_name||m.module)+' <span style="color:var(--muted);font-size:10px;">'+pct+'%</span></span>';
        html += '<span style="color:var(--text);font-variant-numeric:tabular-nums;">↓'+fmtBytes(m.in_bytes)+' ↑'+fmtBytes(m.out_bytes)+'</span>';
      }});
      html += '</div>';
    }} else {{
      html += '<div style="margin-top:6px;color:var(--muted);">暂无今日流量数据 — 采样器启动后每分钟累计一次。</div>';
    }}

    html += dailyBars(t.daily||[]);

    var procs = (t.top_processes)||[];
    if (procs.length) {{
      var iv = t.interval_s ? Math.round(t.interval_s) : '--';
      html += '<div style="margin:10px 0 4px;"><b style="color:var(--text);">🔍 最近 '+iv+' 秒 Top 进程</b></div>';
      html += '<div style="display:grid;grid-template-columns:1fr auto;gap:3px 12px;">';
      procs.slice(0,8).forEach(function(p) {{
        html += '<span style="color:var(--text2);">'+escHtml(p.module_name||p.module)+' <span style="color:var(--muted);font-size:10px;">pid '+p.pid+'</span></span>';
        html += '<span style="color:var(--text);font-variant-numeric:tabular-nums;">↓'+fmtBytes(p.in_bytes)+' ↑'+fmtBytes(p.out_bytes)+'</span>';
      }});
      html += '</div>';
    }}

    html += '<div style="margin-top:12px;font-size:11px;color:var(--muted);">'+
      '说明：一期为进程级总账，「Web服务+游戏交易」同属 unified_server 进程无法拆分；手机热点下浏览器自身流量不计入。'+
      '</div>';

    document.getElementById('trafficBody').innerHTML = html;
  }}

Step 3: JS — loadStatus() 内触发

把 loadStatus 的

      loadDisk();
      loadPerf();

      loadDisk();
      loadPerf();
      loadTraffic();

  loadStatus();
  loadHost();
  setInterval(loadStatus, 30000);
  setInterval(loadHost, 15000);

  window.loadStatus = loadStatus;
  window.loadHost = loadHost;

  loadStatus();
  loadHost();
  loadTraffic();
  setInterval(loadStatus, 30000);
  setInterval(loadHost, 15000);
  setInterval(loadTraffic, 30000);

  window.loadStatus = loadStatus;
  window.loadHost = loadHost;
  window.loadTraffic = loadTraffic;

env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -m py_compile unified_server.py && echo PY_OK
env -u PYTHONHOME -u PYTHONPATH .venv/bin/python -c "
import unified_server as u
h = u._render_system_page()
assert 'id=\"sysTraffic\"' in h, 'card missing'
assert 'function loadTraffic(' in h, 'loader missing'
assert '{{' not in h and '}}' not in h, 'brace restore failed'
print('HTML_OK len=', len(h))
"

Expected: PY_OK + HTML_OK len= <数字>

> _render_system_page() 必须能无异常渲染，且还原花括号后不再残留 {{/}} —— 这能抓住 .format() 转义错误。

git add unified_server.py
git commit -m "feat(traffic): /system 页面流量监控卡片 + 前端渲染"

Run: env -u PYTHONHOME -u PYTHONPATH bash restart.sh Expected: 服务重启成功，端口 8765 监听，公网隧道保持。

/bin/sleep 70
/usr/bin/curl -s http://127.0.0.1:8765/api/system/traffic | /usr/bin/head -c 1200

Expected: JSON 含 "ok": true、"traffic"，today.modules 至少含 web_daemon；interval_s≈60。

用 Chrome DevTools MCP 打开 https://quant.aohan.net/system（或本地 http://127.0.0.1:8765/system），在页面上下文执行：

() => ({
  card: document.querySelectorAll('#sysTraffic').length,
  bodyChildren: document.getElementById('trafficBody').querySelectorAll('div').length,
  subtitle: document.getElementById('trafficSubtitle').textContent,
  badge: document.getElementById('trafficBadge').textContent,
})

Expected: card === 1、bodyChildren > 0、subtitle 含「今日」、badge 为容量串（如 12.3MB）而非 --。若 bodyChildren === 0 或含「尚未」→ 采样未产生，检查步骤 2 与 logs/ 中 [traffic_monitor] 报错。

在真实抓取/游戏交易运行窗口内，观察「今日流量构成」中非 other 模块占比是否随活动上升；other（系统/其他进程）应存在但不独占 100%。

在 learn/ 下新增一篇专题（learn/2026-09-13-网络流量监控/README.md），记录：nettop 差分原理、PID 文件→模块映射、一期限制（网站+游戏交易合并）、二期计划（应用层埋点：sun_requests / 游戏交易 urllib / Playwright response 事件 / web response）。并在 .claude/CLAUDE.md 的「核心模块速查」表补一行链接。

git add learn/ .claude/CLAUDE.md
git commit -m "docs(traffic): 流量监控专题文档 + 索引更新"
git push

• adata/common/utils/sunrequests.py：在 SunRequests 的响应读取处累计 len(resp.content)，按模块上下文打标。
• 游戏交易 urllib 调用点：包一层 counted_urlopen。
• Playwright：监听 response 事件累计 response.body() 大小（注意 content-length 可能缺失）。
• unified_server Web 响应：在 wfile.write 处累加 len(body)。
• 汇总写入 traffic_history 的同表，module 细化到 web:game_trade、fetch:etf_full 等，从而把一期合并的 web_daemon 拆开。

• traffic_history 表创建成功，30 天自动清理生效
• .venv/bin/python -m unittest tests.adata_test.utils.test_traffic_monitor -v 全绿
• /api/system/traffic 返回 200 且含 today/daily/realtime/top_processes
• /system 页面 #sysTraffic 卡片真实渲染出模块构成与趋势（DOM 实测）
• 无 except Exception: pass 静默吞异常；采样失败有日志
• unified_server.py` 用 SearchReplace 修改，未用 cp/cat 覆盖
• 已发布到 https://quant.aohan.net（非仅本地）