tdx-link-mcp 0.1.0

通达信 AI Agent MCP Server - 让 AI 直接操作通达信

tdx_lib — 通达信量化交易框架

把通达信 HTTP 接口抽象成分层模块，写策略只需关注逻辑本身，不再重复造轮子。

---

架构总览

┌──────────────────────────────────────────────────────────┐
│                    strategies/  策略层                      │
│        用户只写这里: 继承 StrategyBase, 实现 analyze()       │
├──────────────────────────────────────────────────────────┤
│           strategy.py        backtest.py    scheduler.py  │
│           策略生命周期        回测引擎       定时调度        │
├──────────────────────────────────────────────────────────┤
│  data.py      trade.py     portfolio.py   risk.py  grid.py│
│  行情数据     交易下单      板块管理      风控止损  网格交易  │
├──────────────────────────────────────────────────────────┤
│              client.py              utils.py              │
│              HTTP通信               计算工具               │
└──────────────────────────────────────────────────────────┘

设计原则: 策略层只依赖框架层，框架层只依赖服务层，服务层只依赖基础层。换策略不用改基础设施，加功能不用动底层通信。

---

功能清单

模块	功能	状态
client.py	HTTP JSON-RPC 通信、批量并发调用、连通性检测	✅
launcher.py	通达信进程检测/自动拉起/HTTP就绪轮询 (一行调用)	✅
data.py	股票列表、K线、快照、more_info 批量查询、预筛框架	✅
financial.py	官方财务/交易数据9接口 (FN/GP/BK/SC/GO系列), 透传官方	✅
screener.py	通用筛选器(透传官方公式B007判定涨跌停, 取数优化)、涨跌停家数秒查	✅
trade.py	买入/卖出/市价/限价/涨跌停价/隔夜挂单/批量撤单/账户查询	✅
portfolio.py	板块创建/删除/清空/重写、预警信号、客户端跳转	✅
strategy.py	策略基类，串联完整生命周期 (选股→预筛→分析→输出→写板块)	✅
risk.py	止损(固定/移动/ATR)、止盈(固定/移动)、仓位管理(固定/比例/凯利)	✅
grid.py	网格交易(等差/等比)，自动低买高卖	✅
backtest.py	历史回测，绩效统计(胜率/回撤/夏普/盈亏比)	✅
scheduler.py	定时执行策略，支持 HH:MM 和 cron 表达式	✅

---

快速上手

1. 选股策略 — 只写 analyze()

# strategies/my_strategy.py
from tdx_lib import StrategyBase, calc_ma, parse_kline_float

class MyStrategy(StrategyBase):
    name = "我的策略"
    block_code = "MINE"
    min_cap = 50
    max_cap = 500

    def analyze(self, code, kline_data):
        parsed = parse_kline_float(kline_data)
        closes = parsed["Close"]
        if len(closes) < 20:
            return None
        ma5 = calc_ma(closes, 5)
        ma20 = calc_ma(closes, 20)
        if ma5 > ma20 and closes[-1] > ma5:
            return {"code": code, "close": round(closes[-1], 2)}
        return None

运行: python run_strategy.py my_strategy

2. 回测 — 一行跑历史数据

from tdx_lib.backtest import BacktestEngine
from strategies.my_strategy import MyStrategy

engine = BacktestEngine(MyStrategy, initial_capital=1_000_000)
engine.trailing_stop = True
report = engine.run(start_offset=60, stock_limit=100)
engine.print_report()

3. 交易 + 止损止盈

from tdx_lib import TdxClient, TdxTrader, RiskManager, StopLoss

client = TdxClient()
trader = TdxTrader(client)
rm = RiskManager(stop_loss_type=StopLoss.Type.TRAILING, sl_pct=5.0,
                 capital=trader.get_total_asset())

# 买入
volume = rm.calc_volume(price=10.0)
trader.limit_buy("600000", volume, 10.0)
rm.on_buy("600000", 10.0)

# 检查止损
should_exit, exit_price, reason = rm.check_exit("600000", high=10.5, low=9.6)

4. 网格交易

from tdx_lib import GridTrader

grid = GridTrader("600000", lower_bound=8, upper_bound=12,
                  grid_count=20, volume_per_grid=200)
grid.setup(current_price=10.0)

actions = grid.on_price_update(current_price=10.5)
# → [{"action": "sell", "price": 10.5, "volume": 200, ...}]

5. 定时自动运行

from tdx_lib import StrategyScheduler

sched = StrategyScheduler()
sched.add_strategy("my_strategy", time_str="15:30")  # 每天15:30选股
sched.add_task(overnight_orders, time_str="14:50", name="隔夜挂单")
sched.run()

6. 条件筛选 — 涨跌停/涨幅/市值 (两层过滤, 秒级)

from tdx_lib import TdxClient, StockScreener, get_market_limit_count

client = TdxClient()

# 秒查涨跌停家数 (1次请求, ~0.005s)
r = get_market_limit_count(client)
print(f"涨停 {r['limit_up']} 家, 跌停 {r['limit_down']} 家")

# 涨停明细 (粗筛涨幅 + 精筛触及涨停价, 全市场 ~2s)
results = (StockScreener.from_all_a_stocks(client)
           .where(min_pct=9.8).limit_up().run())

# 任意条件: 涨幅 5-10% 且市值 30-300亿
results = (StockScreener.from_all_a_stocks(client)
           .where(min_pct=5.0, max_pct=10.0)
           .where(min_cap=30, max_cap=300).run())

---

目录结构

tdxLinkMcp/
├── tdx_lib/                        # 核心框架
│   ├── __init__.py                 # 统一导出
│   ├── client.py                   # HTTP 客户端
│   ├── utils.py                    # 工具函数
│   ├── data.py                     # 行情数据
│   ├── trade.py                    # 交易接口
│   ├── portfolio.py                # 板块管理
│   ├── risk.py                     # 风险管理
│   ├── grid.py                     # 网格交易
│   ├── strategy.py                 # 策略基类
│   ├── backtest.py                 # 回测引擎
│   └── scheduler.py                # 定时调度
├── strategies/                     # 策略目录
│   ├── first_board_pullback.py     # 首板回调捉妖
│   └── macd_cross.py               # MACD金叉
├── run_strategy.py                 # CLI 运行入口
├── SKILL.md                        # AI 导航文档
└── README.md                       # 本文件

---

运行策略

# 列出所有可用策略
python run_strategy.py

# 运行指定策略
python run_strategy.py first_board_pullback
python run_strategy.py macd_cross

策略文件放入 strategies/ 目录后自动注册，文件名即策略名。

---

依赖

• Python 3.8+
• 通达信客户端 (开启 HTTP 服务)
• 标准库 (无第三方依赖)

---

给 AI 的话

首要原则: 官方入参优先。 用户与 AI 沟通形成的策略/条件, 最终必须是通达信官方接口的入参, 本框架只取数与呈现。详见 .trae/rules/official-first.md (项目铁律, 由 tdx_lib/guard.py 运行时强制)。

如果需要修改、扩展、或添加新功能，请先读 .trae/rules/official-first.md 再读 SKILL.md — 前者是强制铁律, 后者是 API 速查与操作指南。