En la carga de los recursos... Cargando...

Herramientas de gestión de la red

El autor:Programación, Fecha: 2023-04-30 09:03:58
Las etiquetas:

Parámetros ¿Es necesario? Descripción
precio superior No Precio en la frontera de la red
precio más bajo No Precio de la línea inferior
número de red No Número de rejillas.
el intervalo No Las distancias entre redes (equivalentes)
el lado No Apoyo a la transmisiónlongshortNo se completa por defecto.long
Datos obtenidos No Ya hay información de la red, tipo diccionario.
# 创建了一个价格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)

Estructura de datos

Parámetros ¿Es necesario? Descripción
lista de cuadrícula Sí, es cierto. Precio de la red y la información de los pedidos, almacenados con el valor clave, clave como precio, valor como orden id
el intervalo Sí, es cierto.
precio superior Sí, es cierto.
precio más bajo Sí, es cierto.
número de red Sí, es cierto.
el lado Sí, es cierto.
la red_diferencia Sí, es cierto.
Tipo de producto Sí, es cierto. Es igual a menos o igual a menos.

Función

  • el precio de compra más cercano.

    Obtenga los precios de compra más recientes de la red

    Parámetros ¿Es necesario? Descripción
    Precio actual Sí, es cierto. Precio de entrada para encontrar el precio de compra más reciente
  • el precio de venta más cercano (precio actual)

    Obtener el precio de venta más reciente de la red

    Parámetros ¿Es necesario? Descripción
    Precio actual Sí, es cierto. Precio de entrada para encontrar el precio de venta más reciente
  • Base_posición (ticker)

    El basamento

    Parámetros ¿Es necesario? Descripción
    el ticker Sí, es cierto. Abre el depósito final para abrir la rejilla, y la función ejecuta la función de recuperacióneventEl hechobase_position
  • Adjunto_orden (orden_id)

    Añadir las listas en la red

    Parámetros ¿Es necesario? Descripción
    el orden_id Sí, es cierto. La función id de la que se añade el listado de la malla, el depósito o la orden de transacción se encuentra en la malla de la misma y ejecuta la función de recuperación.eventEl hechoadd_order
  • cancelar_orden ((orden_id)

    Se retira el pedido

    Parámetros ¿Es necesario? Descripción
    el orden_id Sí, es cierto. Deshacer la orden especificada, esta función ejecuta la función de recuperacióneventEl hechocancel_order

El hecho

Un evento es una función de llamada específica que se llama durante la ejecución de una función.

gm = GridPriceManager(1000, 800, 10)

# 底仓事件,在调用base_position方法时会触发此事件
@gm.event('base_position')
def base_position(price):
    # 传入最近的网格价格,以此价格作为买入价格参考
    print(price)
    return 123456	# 返回底仓订单,manger将订单记录
El hecho ¿Es necesario? La transmisión Volver hacia atrás
base_posición Sí, es cierto. precio, precio de compra, tipo de float Id de pedido de almacén
¿ Qué es esto? No El precio, el precio de compra en la red, el tipo de dict, el tipo de dict, el tipo de dict. Con el mismo formato de dict que se transmite, corresponde al id de transacción de la red superior y al id de transacción de la red inferior.
cancelar_orden No order_id, especifica el tipo de orden id, int o str para revocar bool, ¿ha sido exitoso?
cambio No lista de cuadrícula El cambio en la información de la red provocó el incidente.

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)

Más.