使用python库实现简单网格策略分享

网格交易是一种经典的量化交易策略，其核心思想是在价格上下预设多个“网格”，当价格触发特定网格时执行买入或卖出操作，以低买高卖赚取区间波动收益。

本文基于python语言实现一个简化的网格策略模型。重点在于策略逻辑和行情数据接入，不涉及真实下单。

策略逻辑简述

我们以美股 AAPL 为例，设定如下规则：

• 以当前价格为中心，构造若干上下价格网格
• 每个网格相差 delta_price
• 初始在下方挂多个买单，在上方挂多个卖单
• 当买单成交，在上方新增一个卖单；反之亦然

步骤一：获取实时价格作为网格中心

# 安装
pip install qos-api

from qos_api import QOSClient
# get key: qos.hk
client = QOSClient(api_key="YOUR_API_KEY")

# 获取当前快照
snapshot = client.get_snapshot(["US:AAPL"])
price = snapshot[0].price  # 当前价格作为网格中心
print(f"当前价格: {price}")

步骤二：生成网格

def generate_grid(center_price, delta, grid_num):
    grids = []
    for i in range(-grid_num, grid_num + 1):
        level_price = round(center_price + i * delta, 2)
        grids.append(level_price)
    return sorted(grids)

grid_prices = generate_grid(center_price=price, delta=1.0, grid_num=5)
print("构造的网格价格:", grid_prices)

步骤三：模拟持仓与挂单记录

我们不接入实际交易所，而是用字典模拟持仓和委托。

position = 0
orders = {}  # 记录挂单状态，格式: price -> "buy"/"sell"
filled_orders = set()

步骤四：监控实时价格并执行交易逻辑

使用 WebSocket 实时行情模拟策略运行逻辑。

import asyncio
# get key: qos.hk
async def run_grid_strategy():
    client = QOSClient(api_key="YOUR_API_KEY")

    def decide_trade(price):
        global position
        for grid in grid_prices:
            if price <= grid and orders.get(grid) != "buy" and grid not in filled_orders:
                print(f"触发买入 @ {grid}")
                position += 1
                orders[grid] = "buy"
                filled_orders.add(grid)
                sell_price = round(grid + 1.0, 2)
                orders[sell_price] = "sell"
                print(f"挂出卖单 @ {sell_price}")
                break
            elif price >= grid and orders.get(grid) != "sell" and grid not in filled_orders:
                print(f"触发卖出 @ {grid}")
                position -= 1
                orders[grid] = "sell"
                filled_orders.add(grid)
                buy_price = round(grid - 1.0, 2)
                orders[buy_price] = "buy"
                print(f"挂出买单 @ {buy_price}")
                break

    async def handle_snapshot(data):
        now_price = data[0].price
        print(f"实时价格: {now_price}")
        decide_trade(now_price)

    client.register_callback("S", handle_snapshot)
    await client.connect_ws()
    await client.subscribe_snapshot(["US:AAPL"])

    while True:
        await asyncio.sleep(1)

asyncio.run(run_grid_strategy())

总结

这个简单网格策略原型：

• 实现了实时订阅行情并基于价格触发交易行为
• 构造了固定间距的价格网格
• 在模拟环境中做了挂单与持仓管理

当然，真实应用中你还需要接入交易所 API 执行下单、管理资金、风控处理等。但作为策略验证原型，这已经是一个不错的起点。

欢迎基于这个逻辑扩展更多功能，比如：

• 动态调整网格宽度
• 多品种同时运行
• 加入止盈止损
• 接入真实下单 API

量化交易重在逻辑清晰、执行稳定，从这个网格策略起步，慢慢可以构建出自己的交易体系。