案例：自动化交易 bot¶

Python bot 走 gRPC 下单，simulate 先跑一周验证，再切 real。

5 分钟从零到能下第一单¶

假设你已经装好 futu-opend + futucli + Python 3.10+。从零开始：

# 1. 凭据一次性存 OS keychain（避免 --login-pwd 泄漏到 ps aux）
futucli set-login-pwd --account 12345678
#   提示：输入密码 → 密码进 Keychain / Secret Service / Credential Manager

# 2. 首次启动，前台跑 SMS 验证（服务端首次会发短信到你绑定手机）
futu-opend --setup-only --login-account 12345678 --platform futunn
#   SMS 到手输验证码，看到 "credentials cached" 即可 Ctrl-C 退出

# 3. 启动网关（无人值守，systemd / Docker / nohup 都行）
futu-opend --login-account 12345678 --platform futunn \
           --rest-port 22222 --grpc-port 33333 \
           --audit-log /var/log/futu/
#   监听：FTAPI(11111) / REST(22222) / gRPC(33333)

# 4. 看账户列表（v1.4.26 修好后能看到所有 broker 的全部 real/sim 账户）
futucli account
#   输出示例（8 列：Acc ID / Card / Env / Broker / Type / Status / Role / Markets）
#   记下要交易的 acc_id

# 5. 确认行情能拉（订阅 + 拉一次 basic quote）
futucli quote HK.00700,HK.09988

# 6. 实盘交易前先解锁（real 环境必做，sim 跳过）
futucli unlock-trade --trade-pwd-md5 <32位小写hex>

# 7. 下第一单（sim 环境练手，推荐先跑这条）
futucli place-order \
  --env simulate --market HK --acc-id <sim_acc_id> \
  --code 00700 --side BUY --qty 100 --price 300 --order-type NORMAL

# 8. 查订单 + 成交
futucli order --env simulate --market HK --acc-id <sim_acc_id>
futucli deal --env simulate --market HK --acc-id <sim_acc_id>

走通到这一步，说明 auth + 交易链路全通，bot 只需把 futucli 换成 gRPC 客户端即可。

v1.4.26 新增的行情分析命令¶

v1.4.26 起 futucli 新增 5 个行情分析命令（MCP 对应 tool 同样可用）：

# 资金流向时间序列（period_type=1 分时，2/3/4/... 日线/周线）
futucli capital-flow HK.00700 --period-type 1

# 资金分布（超大/大/中/小单流入流出）
futucli capital-distribution HK.00700

# 多只股票市场状态（开盘/休市/午休/盘后）
futucli market-state HK.00700,US.AAPL,HK.09988

# 一批股票各自所属板块（行业/概念/地域）
futucli owner-plate HK.00700,US.AAPL

# 期权链（按到期日列出 call/put 合约）
futucli option-chain HK.00700 --begin 2026-05-01 --end 2026-06-30 --option-type all

MCP 里对应 tool 名：futu_get_capital_flow / futu_get_capital_distribution / futu_get_market_state / futu_get_owner_plate / futu_get_option_chain，以及新增的 futu_get_history_kline（带 rehab 控制）+ futu_get_reference （关联证券）。见 MCP 接入 LLM。

以下是进阶部分 —— 当你确认 5 分钟入门走通后再看。

需求¶

bot 跑在独立机器，通过 gRPC 连网关
策略是低频挂单 + 定时撤单，每天最多 100 单，单笔 ≤ 5 万
只做港股
simulate 验证期一周，之后切 real
审计 + 告警 + 熔断必须齐

key 策略¶

两把 key 分环境：

# simulate 验证期
futucli gen-key \
  --id bot-sim \
  --scopes qot:read,acc:read,trade:simulate \
  --allowed-markets HK \
  --max-order-value 50000 \
  --max-daily-value 500000 \
  --max-orders-per-minute 2 \
  --hours-window 09:30-16:00 \
  --expires 14d      # 两周够验证期用

# real 上线
futucli gen-key \
  --id bot-prod \
  --scopes qot:read,acc:read,trade:real \
  --allowed-markets HK \
  --max-order-value 50000 \
  --max-daily-value 500000 \
  --max-orders-per-minute 2 \
  --hours-window 09:30-16:00 \
  --bind-machines fp_bot_host \
  --expires 30d

两把 key 独立统计 —— audit 日志、限额计数、metrics 都按 key_id 区分。

网关 + gRPC¶

futu-opend \
  --login-account 12345678 --login-pwd "$FUTU_PWD" \
  --grpc-port 33333 \
  --grpc-keys-file /etc/futu/keys.json \
  --audit-log /var/log/futu/

TLS 终结走 Caddy / nginx（见部署到生产）。

Python bot 代码框架¶

import grpc
from futu_pb2 import FutuRequest
from futu_pb2_grpc import FutuOpenDStub

# 从 env 读 token
import os
API_KEY = os.environ["BOT_API_KEY"]

# metadata 带 Bearer
metadata_cb = grpc.metadata_call_credentials(
    lambda ctx, cb: cb((("authorization", f"Bearer {API_KEY}"),), None)
)
creds = grpc.composite_channel_credentials(
    grpc.ssl_channel_credentials(), metadata_cb
)
channel = grpc.secure_channel("api.your-domain.com:443", creds)
stub = FutuOpenDStub(channel)

def get_quote(code: str):
    body = encode_get_basic_qot(code)
    resp = stub.Request(FutuRequest(proto_id=3004, body=body))
    return parse_get_basic_qot(resp.body)

def place_order(acc_id: int, code: str, side: int, qty: int, price: float):
    body = encode_place_order(acc_id, code, side, qty, price)
    try:
        resp = stub.Request(FutuRequest(proto_id=2202, body=body))
        return parse_place_order(resp.body)
    except grpc.RpcError as e:
        code = e.code()
        if code == grpc.StatusCode.RESOURCE_EXHAUSTED:
            # 限额触发 —— rate / daily / market / symbol / side / hours / per-order
            # 读 e.details() 里的 reason 分桶决定怎么处理
            reason = e.details()
            if "rate limit" in reason:
                time.sleep(60)      # rate: 等一分钟再重试
                raise TransientError(reason)
            elif "daily value cap" in reason:
                raise StopForToday(reason)   # daily: 当天别再交了
            else:
                raise HardFail(reason)       # market / symbol / side / hours: 策略错了
        elif code == grpc.StatusCode.PERMISSION_DENIED:
            raise ConfigError("key scope 不够")
        raise

# 主循环
while trading_window():
    data = get_quote("00700")
    decision = strategy.decide(data)
    if decision:
        place_order(**decision)

熔断：本地 circuit breaker¶

防护栏之外的自我保护 —— 万一服务端护栏有 bug，本地也挡一道：

import collections, time

class LocalLimiter:
    def __init__(self, max_per_minute=2):
        self.max = max_per_minute
        self.window = collections.deque()

    def try_acquire(self):
        now = time.time()
        while self.window and now - self.window[0] > 60:
            self.window.popleft()
        if len(self.window) >= self.max:
            return False
        self.window.append(now)
        return True

limiter = LocalLimiter(max_per_minute=2)
if not limiter.try_acquire():
    raise TransientError("local rate limit")

Prometheus 告警¶

alerts.yml（bot 侧）

- alert: BotRejectSpike
  expr: rate(futu_auth_events_total{key_id=~"bot-(sim|prod)",outcome="reject"}[5m]) > 0.5
  annotations:
    summary: "bot {{ $labels.key_id }} 近 5 分钟异常 reject 多"

- alert: BotDailyCapNearLimit
  expr: futu_auth_limit_rejects_total{key_id=~"bot-.*",reason="daily"} > 3
  annotations:
    summary: "bot {{ $labels.key_id }} 当日累计超限 {{ $value }} 次"

上线节奏¶

第 1 周：bot-sim + 极低限额 + 无真实资金
 ↓ 全绿
第 2 周：bot-sim + 正常限额 + 对比信号与 real 账户（但不真实下单）
 ↓ 全绿
第 3 周：bot-prod + 5% 资金 + 紧限额
 ↓ 观察 2 周
第 5 周：bot-prod + 正常资金 + 正常限额

每一步都有：audit 对比、metrics 核对、策略回放。别跳步。