FMZ量化交易Python版简单网格策略

Python版简单网格策略

策略广场上的Python策略不多，这里编写了一个Python版本的网格策略。策略原理十分简单，在一个价格区间内固定价格距离产生一系列的网格节点，当行情变化时，价格到达一个网格节点价格位置，就挂一个买入订单。当这个订单成交时，即按照挂单的价格加上利润差价，挂出平仓的卖单订单。捕捉在设置的价格区间内的波动。

网格策略的风险不用多说，任何网格类型的策略都是属于赌价格在某个区间震荡，一旦价格冲出网格范围，可能造成严重浮亏。所以写该策略的目的在于，提供Python策略编写思路上或者程序设计上的参考。该策略仅仅用于学习，实盘可能有很大风险。

策略思路讲解直接写在了策略代码注释上。

策略代码

'''backtest
start: 2019-07-01 00:00:00
end: 2020-01-03 00:00:00
period: 1m
exchanges: [{"eid":"OKEX","currency":"BTC_USDT"}]
'''

import json

# 参数
beginPrice = 5000   # 网格区间开始价格
endPrice = 8000     # 网格区间结束价格
distance = 20       # 每个网格节点的价格距离
pointProfit = 50    # 每个网格节点的利润差价
amount = 0.01       # 每个网格节点的挂单量
minBalance = 300    # 账户最小资金余额（买入时）

# 全局变量
arrNet = []
arrMsg = []
acc = None

def findOrder (orderId, NumOfTimes, ordersList = []) :
    for j in range(NumOfTimes) :
        orders = None
        if len(ordersList) == 0:
            orders = _C(exchange.GetOrders)
        else :
            orders = ordersList
        for i in range(len(orders)):
            if orderId == orders[i]["Id"]:
                return True
        Sleep(1000)
    return False

def cancelOrder (price, orderType) :
    orders = _C(exchange.GetOrders)
    for i in range(len(orders)) :
        if price == orders[i]["Price"] and orderType == orders[i]["Type"]:
            exchange.CancelOrder(orders[i]["Id"])
            Sleep(500)

def checkOpenOrders (orders, ticker) :
    global arrNet, arrMsg
    for i in range(len(arrNet)) :
        if not findOrder(arrNet[i]["id"], 1, orders) and arrNet[i]["state"] == "pending" :
            orderId = exchange.Sell(arrNet[i]["coverPrice"], arrNet[i]["amount"], arrNet[i], ticker)
            if orderId :
                arrNet[i]["state"] = "cover"
                arrNet[i]["id"] = orderId
            else :
                # 撤销
                cancelOrder(arrNet[i]["coverPrice"], ORDER_TYPE_SELL)
                arrMsg.append("挂单失败!" + json.dumps(arrNet[i]) + ", time:" + _D())

def checkCoverOrders (orders, ticker) :
    global arrNet, arrMsg
    for i in range(len(arrNet)) :
        if not findOrder(arrNet[i]["id"], 1, orders) and arrNet[i]["state"] == "cover" :
            arrNet[i]["id"] = -1
            arrNet[i]["state"] = "idle"
            Log(arrNet[i], "节点平仓，重置为空闲状态。", "#FF0000")


def onTick () :
    global arrNet, arrMsg, acc

    ticker = _C(exchange.GetTicker)    # 每次获取当前最新的行情
    for i in range(len(arrNet)):       # 遍历所有网格节点，根据当前行情，找出需要挂单的位置，挂买单。
        if i != len(arrNet) - 1 and arrNet[i]["state"] == "idle" and ticker.Sell > arrNet[i]["price"] and ticker.Sell < arrNet[i + 1]["price"]:
            acc = _C(exchange.GetAccount)
            if acc.Balance < minBalance :     # 如果钱不够了，只能跳出，什么都不做了。
                arrMsg.append("资金不足" + json.dumps(acc) + "！" + ", time:" + _D())
                break

            orderId = exchange.Buy(arrNet[i]["price"], arrNet[i]["amount"], arrNet[i], ticker) # 挂买单
            if orderId :
                arrNet[i]["state"] = "pending"   # 如果买单挂单成功，更新网格节点状态等信息
                arrNet[i]["id"] = orderId
            else :
                # 撤单
                cancelOrder(arrNet[i]["price"], ORDER_TYPE_BUY)    # 使用撤单函数撤单
                arrMsg.append("挂单失败!" + json.dumps(arrNet[i]) + ", time:" + _D())
    Sleep(1000)
    orders = _C(exchange.GetOrders)
    checkOpenOrders(orders, ticker)    # 检测所有买单的状态，根据变化做出处理。
    Sleep(1000)
    orders = _C(exchange.GetOrders)
    checkCoverOrders(orders, ticker)   # 检测所有卖单的状态，根据变化做出处理。

    # 以下为构造状态栏信息，可以查看FMZ API 文档。
    tbl = {
        "type" : "table",
        "title" : "网格状态",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

    for i in range(len(arrNet)) :
        tbl["rows"].append([i, json.dumps(arrNet[i])])

    errTbl = {
        "type" : "table",
        "title" : "记录",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

    orderTbl = {
     	"type" : "table",
        "title" : "orders",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

    while len(arrMsg) > 20 :
        arrMsg.pop(0)

    for i in range(len(arrMsg)) :
        errTbl["rows"].append([i, json.dumps(arrMsg[i])])

    for i in range(len(orders)) :
        orderTbl["rows"].append([i, json.dumps(orders[i])])

    LogStatus(_D(), "\n", acc, "\n", "arrMsg length:", len(arrMsg), "\n", "`" + json.dumps([tbl, errTbl, orderTbl]) + "`")


def main ():         # 策略执行从这里开始
    global arrNet
    for i in range(int((endPrice - beginPrice) / distance)):        # for 这个循环根据参数构造了网格的数据结构，是一个列表，储存每个网格节点，每个网格节点的信息如下：
        arrNet.append({
            "price" : beginPrice + i * distance,                    # 该节点的价格
            "amount" : amount,                                      # 订单数量
            "state" : "idle",    # pending / cover / idle           # 节点状态
            "coverPrice" : beginPrice + i * distance + pointProfit, # 节点平仓价格
            "id" : -1,                                              # 节点当前相关的订单的ID
        })

    while True:    # 构造好网格数据结构后，进入策略主要循环
        onTick()   # 主循环上的处理函数，主要处理逻辑
        Sleep(500) # 控制轮询频率

策略主要设计思路是，根据自己维护的这个网格数据结构，对比GetOrders接口返回的当前挂单列表。分析挂出的订单变化（成交与否），更新网格数据结构，做出后续操作。并且挂出的订单不会撤销，直到成交，即使价格偏离也不撤销，因为数字货币市场经常有插针的情况，这些挂单也有可能接到插针的单子（如果交易所有挂单数量限制，那就要调整了）。

策略数据可视化，使用了LogStatus函数把数据实时显示在状态栏上。

    tbl = {
        "type" : "table",
        "title" : "网格状态",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

    for i in range(len(arrNet)) :
        tbl["rows"].append([i, json.dumps(arrNet[i])])

    errTbl = {
        "type" : "table",
        "title" : "记录",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

    orderTbl = {
     	"type" : "table",
        "title" : "orders",
        "cols" : ["节点索引", "详细信息"],
        "rows" : [],
    }

构造了三个表格，第一个表格显示当前网格数据结构中每个节点的信息，第二个表格显示异常信息，第三个表格显示交易所实际挂单信息。

回测测试

策略地址

策略地址

策略仅供参考学习，回测测试，有兴趣可以优化升级。