网格管理工具

Author: program, Date: 2023-04-30 09:03:58
Tags:

参数 是否必需 描述
upper_price NO 网格上边界价格
lower_price NO 网格下边界价格
grid_num NO 网格数量(等差)
interval NO 网格间隔(等比)
side NO 支持传入longshort,不填写则默认long
Data NO 已有网格信息,类型为字典
# 创建了一个价格1000-800,数量为10的等差网格
GridPriceManager(upper_price=1000, lower_price=800, grid_num=10)

# 创建了一个价格1000-800,间隔为1%的等比网格
GridPriceManager(upper_price=1000, lower_price=800, interval=1)

# 传入已有网格信息
data = {
	"grid_list":    {99:None,100:None,101:None,102:None,103:None,104:None},
	"interval":     None,
	"upper_price":  104,
	"lower_price":  99,
	"grid_num":     6,
	"side":         "long",
	"grid_diff":    1,
	"type":         "等差",
}
GridPriceManager(Data=data)

Data结构

参数 是否必需 描述
grid_list YES 网格价格与订单信息,以键值对储存,key为价格,value为订单id
interval YES
upper_price YES
lower_price YES
grid_num YES
side YES
grid_diff YES
type YES 等差 or 等比

函数

  • get_nearest_buy_price(current_price)

    获取最近网格买入价格

    参数 是否必需 描述
    current_price YES 传入价格,以此价格找到最近的买入价格
  • get_nearest_sell_price(current_price)

    获取最近网格卖出价格

    参数 是否必需 描述
    current_price YES 传入价格,以此价格找到最近的卖出价格
  • base_position(ticker)

    底仓

    参数 是否必需 描述
    ticker YES 开底仓,以此开启网格,此函数会执行回调函数event事件base_position
  • add_order(order_id)

    增加网格上下挂单

    参数 是否必需 描述
    order_id YES 增加网格上下挂单,传入底仓或成交订单的id函数将找到此id的上下网格,此函数会执行回调函数event事件add_order
  • cancel_order(order_id)

    撤销订单

    参数 是否必需 描述
    order_id YES 撤销指定订单,此函数会执行回调函数event事件cancel_order

事件

事件是指函数执行过程中调用的指定回调函数,这里一律使用event函数传入指定事件,装饰器模式

gm = GridPriceManager(1000, 800, 10)

# 底仓事件,在调用base_position方法时会触发此事件
@gm.event('base_position')
def base_position(price):
    # 传入最近的网格价格,以此价格作为买入价格参考
    print(price)
    return 123456	# 返回底仓订单,manger将订单记录
事件 是否必需 传入 返回
base_position YES price,买入价格,float类型 底仓订单id
add_order NO price,买入网格价格,dict类型,{“up”:上网格价格,“down”:下网格价格} 与传入同格式的dict,对应上网格成交id,下网格成交id
cancel_order NO order_id,指定撤销的订单id,int或str类型 bool,是否撤销成功
change NO grid_list 网格信息发生变化触发此事件

class GridPriceManager:
    def __init__(self, Data=None, upper_price=None, lower_price=None, interval=None, grid_num=None, side: Literal['long','short']='long') -> dict:
        self.interval = interval
        self.upper_price = upper_price
        self.lower_price = lower_price
        self.grid_num = grid_num
        self.side = side
        self.grid_diff = None
        self.type = None    # 网格类型
        if self.grid_num is not None:
            self.grid_diff = (self.upper_price - self.lower_price) / (self.grid_num - 1)
        if Data is None: 
            if self.interval is None:
                self.grid_list = self._generate_grid_list_difference()
                self.type = "等差"
            else:
                self.grid_list = self._generate_grids_list_ratio()
                self.type = "等比"
        else:
            self.grid_list = Data["grid_list"]
            self.interval = Data["interval"]
            self.upper_price = Data["upper_price"]
            self.lower_price = Data["lower_price"]
            self.grid_num = Data["grid_num"]
            self.side = Data["side"]
            self.grid_diff = Data["grid_diff"]
            self.type = Data["type"]
        self.data = f"网格类型: {self.type}, 网格数量: {len(self.grid_list)}, 上下区间: [{self.upper_price}-{self.lower_price}, 方向: {self.side}]"
        self.callback = {}

    def event(self, event_name):
        """事件"""
        def decorator(func):
            self.callback[event_name] = func
            return func
        return decorator

    def _generate_grid_list_difference(self) -> dict:
        """等差网格生成"""
        grid_list = {}
        price = self.lower_price
        for _ in range(self.grid_num):
            grid_list[price] = None
            price += self.grid_diff
        grid_list[self.upper_price] = None
        return grid_list

    def _generate_grids_list_ratio(self) -> dict:
        """等比网格生成"""
        ratio = 1 + self.interval / 100
        grid = [self.lower_price * (ratio ** i) for i in range(-100, 101)]
        return {round(g, 8): None for g in grid if self.lower_price <= g <= self.upper_price}


    def get_nearest_buy_price(self, current_price) -> float:
        """获取最近网格买入价格"""
        nearest_price = None
        for price in sorted(self.grid_list.keys()):
            if price > current_price:
                break
            nearest_price = price
        return nearest_price

    def get_nearest_sell_price(self, current_price) -> float:
        """获取最近网格卖出价格"""
        nearest_price = None
        for price in sorted(self.grid_list.keys(), reverse=True):
            if price < current_price:
                break
            nearest_price = price
        return nearest_price
    
    def base_position(self, ticker) -> Union[str, int]:
        """底仓"""
        if self.side == "short":
            t = self.get_nearest_sell_price(ticker)
        else:
            t = self.get_nearest_buy_price(ticker)
        order_id = self.callback["base_position"](t)
        self.grid_list[t] = order_id
        self.callback["change"](self.grid_list)
        return order_id
    
    def add_order(self, order_id) -> Union[Dict, bool]:
        """增加网格上下挂单"""
        up_price = None
        down_price = None
        ticker = None
        keys = list(self.grid_list.keys())
        for i in range(len(keys)-1):
            if self.grid_list[keys[i]] == order_id:
                ticker = keys[i]
                try:
                    if self.side is None or self.side == "long":
                        up_price = keys[i+1]
                        down_price = keys[i-1]
                    else:
                        up_price = keys[i-1]
                        down_price = keys[i+1]
                except IndexError:
                    return False
                break

        PriceDict = {"up": up_price, "down": down_price}
        d = self.callback["add_order"](PriceDict)
        d = {"up": d["up"], "down": d["down"]}
        self.grid_list[up_price] = d["up"]
        self.grid_list[down_price] = d["down"]
        self.grid_list[ticker] = None
        self.callback["change"](self.grid_list)
        return d
    
    def cancel_order(self, order_id):
        """撤销订单"""
        result = self.callback["cancel_order"](order_id)
        if result == True:
            for items in self.grid_list.items():
                if items[1] == order_id:
                    self.grid_list[items[0]] = None
                    self.callback["change"](self.grid_list)
                    break

def main():
    gm = GridPriceManager(1000, 500, 10)

    @gm.event('add_order')
    def add_order(price):
        print(price)
        return {
            'up': 36543,
            'down': 87957,
        }

    @gm.event('cancel_order')
    def cancel_order(order_id):
        return True

    @gm.event('base_position')
    def base_position(price):
        print(price)
        return 123456

    a = gm.base_position(600)
    print(a)
    a = gm.add_order(123456)
    print(gm.grid_list)
    gm.cancel_order(87957)
    print(gm.grid_list)

More