sdmarket 0.1.1

Python SDK for real-time market data, analysis, and trading assistance

SD-API

Python SDK for real-time market data, analysis, and trading assistance

---

项目简介

SD-API 是一个面向 量化交易 场景的 Python 客户端 SDK，旨在为交易者、策略工程师和数据分析师提供一套简洁、统一、可持续扩展的工具链。

它专注于 消费 行情 API（REST / SSE），而不是构建服务端。其核心职责包括：

• 实时行情获取（单产品 / 全市场）
• 历史数据查询与本地存储
• 行情可视化（K 线、点差、延迟）
• 量化指标计算（SMA、EMA、spread）
• 交易信号生成与模拟执行
• 轻量策略框架

为什么要做这个库？

在量化交易开发中，往往需要对接多个行情源、做大量的数据清洗、绘图、指标计算和信号判断。市面上的库要么过于简单（只封装 HTTP），要么过于复杂（绑定特定券商）。SD-API 的定位是：轻量可组合，专注行情侧，不绑定交易层。

安装方式

# 基础安装
pip install sdmarket

# 本地开发安装
cd sd-api
pip install -e .

# 带全部可选依赖
pip install sdmarket[all]

# 分组安装
pip install sdmarket[storage]    # 数据存储
pip install sdmarket[plot]       # 绘图
pip install sdmarket[ta]         # 技术分析
pip install sdmarket[dev]        # 开发测试

快速开始

初始化客户端

from sd_api import SDClient, SDStream

client = SDClient(base_url="http://127.0.0.1:5000")

健康检查

health = client.system.health()
print(health.status, health.server_time_shanghai)

产品列表

products = client.system.products()
print(products["XAUUSD"])  # {'size': 100, 'lev': 500, 'currency': 'USD', 'type': 'metal'}

实时单产品示例

quote = client.quote.live("XAUUSD")
print(quote.symbol, quote.bid, quote.ask, quote.spread, quote.latency_ms)

实时全市场示例

snapshots = client.quote.live_all()
for q in snapshots.data:
    print(q.symbol, q.bid, q.ask)

历史数据示例

from datetime import datetime, timedelta

# 最近 1 小时
hist = client.hist.quote_history("XAUUSD", limit=500)
print(hist.symbol, len(hist.data))

K 线示例

kline = client.kline.get("XAUUSD", tf="5min", limit=100)
for bar in kline.bars:
    print(bar.ts, bar.open, bar.high, bar.low, bar.close)

SSE 监听示例

from sd_api import SDStream

stream = SDStream(base_url="http://127.0.0.1:5000")

# 单产品实时行情
for quote in stream.quote("XAUUSD"):
    print(quote.symbol, quote.bid, quote.ask, quote.latency_ms)
    # 业务逻辑：判断是否要下单、存储、更新 UI 等

SSE 全市场快照

for snapshot in stream.quotes():
    print(snapshot.server_time_shanghai, snapshot.count)

Callback 模式

from sd_api.stream import Dispatcher

dispatcher = Dispatcher()

@dispatcher.on_quote
def handle_quote(quote):
    print(f"[{quote.symbol}] bid={quote.bid} ask={quote.ask} latency={quote.latency_ms}ms")

for snapshot in stream.quotes(callback=dispatcher):
    pass  # dispatcher 自动分发到注册的 handler

本地存储示例

from sd_api.storage import CSVStore, SQLiteStore

# CSV 存储
store = CSVStore(base_dir="./data")
store.save_quote(quote)
store.save_quotes([quote, quote])

# SQLite 存储
store = SQLiteStore(db_path="./data/market.db")
store.save_kline(kline.bars)
bars = store.load_kline("XAUUSD", tf="5min", limit=100)

绘图示例

import matplotlib
matplotlib.use("Agg")  # 非交互式后端
import matplotlib.pyplot as plt

from sd_api.plot import plot_kline, plot_spread, plot_latency

# K 线图
fig, ax = plt.subplots()
plot_kline(kline.bars, ax=ax)
plt.savefig("kline.png")

# 点差序列图
plot_spread(hist.to_pandas())
plt.savefig("spread.png")

# 延迟图
plot_latency(hist.to_pandas())
plt.savefig("latency.png")

简单分析示例

from sd_api.analysis import sma, ema, latency_summary, spread_stats, crossover_signal

# 计算均线
df = hist.to_pandas()
df["sma20"] = sma(df["close"], 20)
df["ema10"] = ema(df["close"], 10)

# 延迟统计
stats = latency_summary(hist.data)
print(stats.latency_mean, stats.latency_max, stats.stale_ratio)

# 点差统计
s_stats = spread_stats(hist.data)
print(s_stats.spread_mean, s_stats.spread_max)

# 交叉信号
signal = crossover_signal(df["close"], fast=10, slow=20)
print(signal[-5:])

交易辅助示例

from sd_api.trading import SimulatedExecutor, TradeSignal, OrderRequest
from sd_api.analysis import crossover_signal

executor = SimulatedExecutor()

# 基于信号发出交易指令
signal = TradeSignal(
    symbol="XAUUSD",
    side="BUY",
    reason="SMA crossover",
    confidence=0.85,
    price=2050.5,
)
feedback = executor.send_signal(signal)
print(feedback.ok, feedback.message)

策略 Runner 示例

from sd_api import SDClient, SDStream
from sd_api.strategy import BaseStrategy, StrategyRunner, StrategyContext

class DemoStrategy(BaseStrategy):
    def on_quote(self, quote: "Quote", ctx: StrategyContext) -> None:
        if quote.symbol == "XAUUSD" and quote.latency_ms is not None:
            if quote.latency_ms < 500:
                print(f"[{quote.symbol}] 低延迟: {quote.latency_ms}ms")
            else:
                print(f"[{quote.symbol}] 延迟警告: {quote.latency_ms}ms")

client = SDClient(base_url="http://127.0.0.1:5000")
stream = SDStream(base_url="http://127.0.0.1:5000")

runner = StrategyRunner(client=client, stream=stream, strategy=DemoStrategy())
runner.run_quote("XAUUSD")

项目路线图

• v0.1.0 — 核心客户端（REST + SSE 消费）
• v0.1.0 — 存储层（CSV / SQLite / Parquet）
• v0.1.0 — 分析层（SMA、EMA、spread、latency）
• v0.1.0 — 绘图层（K 线、点差、延迟）
• v0.1.0 — 交易辅助（信号、执行器、风险）
• v0.1.0 — 策略框架
• v0.2.0 — 策略回测模块
• v0.2.0 — 真实交易执行层
• v0.3.0 — 风控引擎
• v0.3.0 — 多数据源聚合

许可证

MIT License