Grid management tools

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

Parameters Is it necessary? Describe
upper_price NO The price of the border on the net
lower_price NO Price at the bottom of the grid
grid_num NO The number of grids (i.e. different)
interval NO Grid spacing (equivalent)
side NO Support for transmissionlongshortIf you don't fill it in, it's by default.long
Data NO There is already grid information, type in dictionary
# 创建了一个价格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)

The data structure

Parameters Is it necessary? Describe
grid_list YES Grid price and order information stored as key, key as price, value as order id
interval YES
upper_price YES
lower_price YES
grid_num YES
side YES
grid_diff YES
type YES Equal to or equal to

Function

  • get_nearest_buy_price(current_price)

    Get the most recent net purchase price

    Parameters Is it necessary? Describe
    current_price YES Enter the price to find the most recent purchase price
  • get_nearest_sell_price(current_price)

    Get the most recent grid sale price

    Parameters Is it necessary? Describe
    current_price YES Enter the price to find the most recent sale price.
  • base_position(ticker)

    The basement

    Parameters Is it necessary? Describe
    ticker YES Open the bottom shell to open the grid, which executes the callback functioneventThe incidentbase_position
  • add_order(order_id)

    Adding a listing to the grid

    Parameters Is it necessary? Describe
    order_id YES The id function for adding a grid-up listing, passing a bottom-up or transaction order will find the up-down grid for this id, which will execute the callback function.eventThe incidentadd_order
  • cancel_order(order_id)

    Order withdrawal

    Parameters Is it necessary? Describe
    order_id YES Cancel specified order, this function performs callback functioneventThe incidentcancel_order

The incident

An event is a specified callback function that is called during the execution of a function.

gm = GridPriceManager(1000, 800, 10)

# 底仓事件,在调用base_position方法时会触发此事件
@gm.event('base_position')
def base_position(price):
    # 传入最近的网格价格,以此价格作为买入价格参考
    print(price)
    return 123456	# 返回底仓订单,manger将订单记录
The incident Is it necessary? Infiltration Going back
base_position YES price, purchase price, float type Order id of the warehouse
add_order NO price, buy into grid price, type of dict, {up tip: up grid price, down tip: down grid price} Dict, corresponding to upper grid transaction id, lower grid transaction id, is transmitted in the same format
cancel_order NO order_id, specify the type of order id, int or str to cancel bool, whether to undo successfully
change NO grid_list The incident was triggered by a change in grid information.

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