1. Introduction
When hummingbot-api, an MCP server, and Codex CLI are already deployed, the remaining gap is orchestration. You need a lightweight loop that restores state, asks Codex for a strict JSON decision, and passes it to your executor. This article uses a practical pattern. A single Python script runs every 30 minutes, injects previous_state into the prompt, and writes one JSON line as the new state. In my setup, I overwrite the state file on purpose. That keeps only the latest snapshot.
当你已经部署好 hummingbot-api、MCP 服务器和 Codex CLI 后，剩下的关键就是“编排层”。你需要一个轻量循环：恢复上一轮状态、让 Codex 输出严格 JSON 决策、再交给执行层去下单/平仓。本文展示一个实用模式。用一个 Python 脚本每 30 分钟运行一次。它把 previous_state 注入提示词，并写回一条 JSON 作为新状态。我这里是有意覆盖文件。这样只保留最新状态快照，方便查看与回滚。
2. Preparation:
Before running the script, confirm you have Python 3.9+. Make sure the codex command is available in your PATH. Also confirm you can write to ~/trading_plans/. The script uses only the standard library. To sanity-check the CLI, run codex --help. You should see the usage output.
运行脚本前，先确认本机有 Python 3.9+。确保终端能直接调用 codex（已在 PATH 里）。还要确认对 ~/trading_plans/ 有写入权限。脚本只用标准库，不需要额外 pip 包。你可以执行 codex --help 做自检。能正常输出帮助信息就说明 CLI 没问题。
3. Run It:
To run, copy your script into a file such as run.py. Keep LOG_PATH pointing to your preferred location. Then execute python3 run.py. The script reads the previous state, or “null” on the first run. It builds the prompt and calls Codex. It prints part of the raw output to the console. Finally, it writes the result to the state file. Because it sleeps for 30 minutes, you can run it in screen/tmux. You can also run it as a service.
运行步骤很简单。把代码复制到 run.py。把 LOG_PATH 指向你想要的状态文件位置。然后执行 python3 run.py。脚本会读取上一轮状态；首次运行是 “null”。它会构造 prompt 并调用 Codex。控制台会打印一部分原始输出，方便你检查。最后它把结果写回状态文件。脚本每 30 分钟循环一次。你可以放在 screen/tmux 里长期跑。也可以做成 systemd 服务。
#!/usr/bin/env python3
import subprocess
import datetime
import os
import json
from pathlib import Path
# 日志文件：一行一个 JSON，方便后续排查/分析
LOG_PATH = Path.home() / "trading_plans" / "btc_hourly_state.jsonl"
PROMPT_TEMPLATE = """
你现在处于一个循环执行的自动交易环境中，每次运行都会收到一个名为 previous_state 的字符串。
你主要与hummingbot的MCP服务器交互
你现在变成了每30分钟进行一次检查
记得读取当前仓位和余额情况
这个 previous_state 字符串的内容，是上一轮运行时产生并写入日志文件（btc_hourly_state.jsonl）的一行 JSON 状态，或者是字符串 "null"（表示目前没有历史状态）。
本系统的核心目标：
1. 通过多维度行情分析（包括币种选择、近期 K 线结构、波动率、趋势与情绪/恐慌程度等），对合适的交易标的做出方向与仓位决策；
2. 自动管理仓位与风险（包含连续失败次数、风控收缩、暂缓交易等）；
3. 每一轮输出一条完整的 JSON 状态，作为闭环策略的一部分，被写入日志并作为下一轮的输入。
====================【状态恢复要求：previous_state】====================
在生成本轮决策前，你必须先根据 previous_state 恢复上一次的内部状态：
- 如果 previous_state 是字符串 "null"，或者内容不是合法 JSON：
- 说明这是第一次运行或历史日志损坏；
- 你需要假定：
- 当前仓位为空（side 为 "none"，size 为 0）；
- retry_count 为 0；
- 上一轮没有未平仓的订单。
- 如果 previous_state 是合法的 JSON：
- 尽量从中读取上一轮的状态字段，比如：
- 用 "position_after" 作为本轮的初始仓位（position_before）；
- 用 "risk.retry_count" 作为当前的连续失败次数；
- 必要时参考上一轮 "orders" 列表中的订单结果，用来修正你对当前仓位的判断；
- 若上一轮有 "symbol"，可将其视为本轮默认继续关注的主交易对。
- 如果某些字段不存在或为 null，要自动用安全的默认值替代，不要报错，不要中断执行。
- 如果 没有发现余额,请重新尝试,因为大概率是会有的
====================【币种选择与行情分析要求】====================
本系统支持针对多个潜在交易标的进行评估，然后选出一个“本轮重点操作的币种”，例如：
- BTCUSDT（默认主交易对）
- ETHUSDT
- SOLUSDT
- 其他你认为适合作为示例的合约交易标的（可以自行扩展）。
你需要在内部完成以下步骤（这些分析过程不用逐条输出，只需体现在最终 JSON 的字段中）：
1. 币种候选集合评估：
- 根据你能获取到的行情数据（例如通过本地交易执行服务 hummingbot-api / Gate.io 接口）对几个候选币种进行基本评估：
- 最近 24 小时或至少最近若干小时的波动率（例如 1 小时 K 线）；
- 当前深度与成交活跃度（尽量优先选择流动性较好的品种）；
- 是否存在明显趋势（上涨/下跌）或区间震荡。
2. 近期 K 线及趋势分析（对你选定的主交易对）：
- 重点分析你最终选择的 symbol（例如 BTCUSDT）在 **最近若干根 1 小时 K 线** 或其他你认为合理的周期上的结构：
- 判断当前是上升趋势、下降趋势还是震荡行情；
- 粗略判断支撑/压力区域（可基于近期高低点）；
- 估计当前波动率水平（high / medium / low）。
3. 恐慌/情绪分析（可以是主观建模）：
- 你可以基于以下信息推断一个“情绪/恐慌程度”：
- 突发的大幅波动（单根 K 线异常放量或长影线）；
- 很高或很低的波动率；
- 若你能访问到资金费率、未平仓合约量（Open Interest）等信息，也可作为参考；
- 最终总结一个情绪等级，例如：
- "fear"（恐慌偏空）
- "greed"（贪婪偏多）
- "neutral"（中性）
4. 结合行情 + 情绪得出交易方向：
- 综合趋势、波动率、情绪和 previous_state 中的风控信息（例如 retry_count 是否过高）：
- 若趋势向上、情绪偏多或中性，且风险情况允许，可以考虑 "long"；
- 若趋势向下、情绪偏空，且无连续失败重压，可以考虑 "short"；
- 若行情极度混乱或连续多轮失败，可考虑 "close" 或保持观望（不加新仓，甚至主动减仓）。
====================【交易执行与风控要求】====================
关于交易执行：
- 使用MCP服务器来下单,最好再gate的永续合约下单,永续合约和现货是统一账户,现货账户的金额就是永续的金额
- 假定系统中存在一个本地的交易执行服务（例如 hummingbot-api），并提供 HTTP 接口读取行情、获取余额、下单和撤单。
- 默认优先使用 Gate.io 的合约（perpetual futures）下单：
- 可以根据行情选择开多（long）或开空（short）；
- 平仓时可以使用 reduce_only 模式。
- 若在某一轮中无法成功调用外部交易接口：
- 仍然要生成一条合理的 JSON 状态（例如标记本轮下单失败、retry_count 增加），不要让整个任务中断。
- 你要敢于下单,同时要注意风险.
风控逻辑（你需要在内部考虑，并体现在 risk 字段和 position_after 中）：
- 如果上一轮或最近几轮连续失败（retry_count 较大）：
- 本轮应适当减少仓位规模或选择不开新仓。
- 如果当前波动率评估为 "high"：
- 可以选择减仓、降低杠杆或提高风控等级。
- 如果余额不足或仓位过大：
- 不应再继续加仓，优先考虑平部分仓位或观望。
- 注意设置止盈止损
====================【本轮 JSON 输出结构要求】====================
你本轮输出的 JSON 顶层必须是一个对象，只能输出这一段 JSON，不要输出任何额外的文字。
推荐包含如下字段（字段名用双引号表示，这里仅列出名称和含义）：
- "timestamp"：本轮决策的时间戳（ISO8601 字符串，例如 "2025-12-02T12:34:56"）。
- "symbol"：你本轮最终选择的主交易对名称，例如 "BTCUSDT"、"ETHUSDT" 等。
- "candidate_symbols"：一个字符串数组，记录本轮评估过的候选币种列表（例如 ["BTCUSDT","ETHUSDT","PEPEUSDT"]）。
- "price"：你在本轮做决策时参考的主交易对价格（数字）。
- "timeframe"：本轮主要参考的 K 线周期，例如 "1h"、"4h" 等。
- "trend_signal"：你的方向信号，取值限定为 "long"、"short" 或 "close"。
- "emotion"：你对市场情绪/恐慌程度的总结，例如 "fear"、"greed" 或 "neutral"。
- "position_before"：一个对象，表示你推断的本轮开始前的仓位状态，包含字段：
- "side"：long / short / none。
- "size"：当前仓位张数或币的数量（数字）。
- "value"：当前仓位名义价值，单位可以用 USDT（数字）。
- "leverage"：当前使用的杠杆倍数（数字）。
- "orders"：一个数组，每个元素是一个对象，表示本轮你实际尝试执行的订单（可以为空数组），每个订单对象建议包含：
- "action"：本轮的操作类型，限定为 "open_long"、"open_short"、"close_long"、"close_short" 或 "no_action"。
- "order_id"：实际下单返回的订单 ID 字符串，如果没有则为 null。
- "size"：本次操作的数量（数字）。
- "status"：本次订单状态，限定为 "submitted"、"filled"、"failed" 或 "canceled"。
- "risk"：一个对象，记录风险和运行状态，包含字段：
- "volatility"：你对当前波动的判断，限定为 "high"、"medium" 或 "low"。
- "retry_count"：最近连续失败的尝试次数（整数）。如果上一轮失败，本轮在上一轮基础上加一；如果本轮成功，可以重置为 0。
- "position_after"：一个对象，表示你认为在本轮操作之后的仓位状态，与 position_before 字段结构相同。
- "meta"：一个对象，用于附加信息，例如：
- "note"：对本轮行为的简短中文总结（可以简要说明：为什么选这个币种、为什么做这个方向、对下一轮的简单预期）。
重要输出规则：
- 你的最终回答必须是一个严格的 JSON 对象文本，不能有任何额外的说明、前缀或后缀。
- JSON 可以是单行或多行，但脚本会整体当成一条 JSON 写入文件，因此不要在 JSON 外输出其他内容。
- 字段名必须使用双引号，字符串值使用双引号，布尔值使用 true/false，空值使用 null，数值使用数字。
下面是上一轮的状态字符串 previous_state（可能是 "null" 或一段 JSON 文本）：
{previous_state}
请你：
1. 先根据 previous_state 恢复状态；
2. 根据多币种评估、K 线与行情走势、恐慌/情绪与风险约束，选择一个主交易对并做出策略决策；
3. 可在内部调用交易接口执行实际操作；
4. 最后输出一条符合上述结构要求的 JSON 状态，作为本轮的日志。
"""
# ===============================================================
def load_last_state() -> str:
"""
读取上一行 JSON 作为 previous_state（字符串形式传给 codex）。
如果没有或损坏，就返回 "null"。
"""
if not LOG_PATH.exists():
return "null"
try:
with open(LOG_PATH, "r", encoding="utf-8") as f:
lines = f.read().strip().splitlines()
if not lines:
return "null"
last_line = lines[-1]
# 简单验证是否是合法 JSON
json.loads(last_line)
return last_line
except Exception:
# 日志坏了就告诉模型「未知」
return "null"
def build_prompt(previous_state: str) -> str:
"""
根据 PROMPT_TEMPLATE 和上一条状态构造最终给 codex 的 prompt。
这里假定 PROMPT_TEMPLATE 里有 {previous_state} 这个占位符。
"""
if not PROMPT_TEMPLATE:
# 没填 prompt 时，给个安全提示，防止跑空
return (
"PROMPT_TEMPLATE 未配置。这是一条占位提示，只用于防止 codex 收到空文本。"
"请在 Python 代码中设置 PROMPT_TEMPLATE。"
)
return PROMPT_TEMPLATE.format(previous_state=previous_state)
def run_codex(prompt: str) -> str:
"""
调用 codex CLI（命令行），把 prompt 丢进去，拿到原始文本输出。
默认用: codex exec "<prompt>"。
如需指定模型，可以在这里加参数。
"""
try:
completed = subprocess.run(
[
"codex",
"exec",
"--skip-git-repo-check",
"--model",
"gpt-5.1-codex-mini",
prompt,
],
capture_output=True,
text=True,
check=False,
)
except FileNotFoundError:
# codex 不存在时，返回一个 error JSON 字符串
return json.dumps({
"error": "codex_not_found",
"message": "找不到 codex 命令，请检查 PATH 或安装 codex。",
}, ensure_ascii=False)
if completed.returncode != 0:
# codex 本身报错，也用 JSON 包一层，避免日志里出现「无法 parse」
return json.dumps({
"error": "codex_exec_failed",
"message": "codex exec 返回非零状态码",
"stdout": completed.stdout,
"stderr": completed.stderr,
}, ensure_ascii=False)
return completed.stdout.strip()
def append_json_line(raw: str) -> None:
"""
把 codex 输出追加写入 JSONL 文件：
- 如果 raw 是合法 JSON，原样写入；
- 否则包装成 {"status": "ERROR", "raw_output": "..."} 再写入。
"""
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
try:
obj = json.loads(raw)
except Exception:
obj = {
"timestamp": datetime.datetime.now().isoformat(),
"status": "ERROR",
"raw_output": raw,
}
with open(LOG_PATH, "w", encoding="utf-8") as f:
f.write(json.dumps(obj, ensure_ascii=False))
f.write("\n")
def main():
# 读取上一条状态
previous_state = load_last_state()
# 构造本次的 prompt
prompt = build_prompt(previous_state)
print("正在调用 codex ...")
raw_output = run_codex(prompt)
# 控制台简单打印前 500 字，方便偶尔手动检查
print("=== codex 原始输出(前 500 字) ===")
print(raw_output[:500])
print("================================")
# 写入 JSONL 日志
append_json_line(raw_output)
print(f"已写入日志: {LOG_PATH}")
if __name__ == "__main__":
import time
while True:
main()
time.sleep(60*30)
4. Prompt Tuning
The best part of this architecture is simple. Strategy iteration becomes prompt iteration. You change behavior by editing PROMPT_TEMPLATE. You do not need to rewrite the orchestration code each time. You can tweak candidate_symbols, risk rules, and decision boundaries. For example, treat high volatility as a no-new-position signal. Keep the output schema stable. Require the same fields every run. Force double-quoted JSON strings. Also require exactly one JSON object, with no commentary. Then your MCP executor can parse and act predictably.
这个架构最大的优势很简单：策略迭代变成“提示词迭代”。你主要改 PROMPT_TEMPLATE，不必频繁改编排代码。你可以调整 candidate_symbols、风控规则和决策边界。比如把“高波动”当作不开新仓的信号。最重要的是保持输出结构稳定。每轮都输出相同字段。字符串统一用双引号。只允许输出一个 JSON 对象，不带解释。这样下游执行器（MCP 控制 Hummingbot）就能稳定解析和执行。你也可以持续迭代逻辑，而不容易把系统弄崩。