Les ressources ont été chargées... Je charge...

Outil de gestion de grille

Auteur:Le programme, Date: 2023-04-30 09:03:58 Je suis désolé
Les étiquettes:

Paramètres Est-ce que cela est nécessaire? Décrire
prix supérieur - Je ne sais pas. Le prix à la frontière du réseau
prix inférieur - Je ne sais pas. Le prix de la grille inférieure
Numéro de grille - Je ne sais pas. Nombre de grilles (différentes)
l'intervalle - Je ne sais pas. L'écart de grille (équivalent)
côté - Je ne sais pas. Aide à la diffusionlongshortLe texte est en anglais.long
Données - Je ne sais pas. Il y a des informations de grille, le type est le dictionnaire
# 创建了一个价格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)

Structure de données

Paramètres Est-ce que cela est nécessaire? Décrire
liste de grille Oui, oui. Les prix du réseau et les informations d'ordre sont stockés avec des valeurs de clé, des prix de clé et des valeurs d'ordre id.
l'intervalle Oui, oui.
prix supérieur Oui, oui.
prix inférieur Oui, oui.
Numéro de grille Oui, oui.
côté Oui, oui.
réseau_diff Oui, oui.
type Oui, oui. L'équivalent ou l'équivalent

Les fonctions

  • le prix d'achat le plus proche (prix actuel)

    Retrouvez les prix d'achat les plus récents

    Paramètres Est-ce que cela est nécessaire? Décrire
    prix actuel Oui, oui. Envoyez le prix pour trouver le prix le plus récent.
  • le prix de vente le plus proche (prix actuel)

    Obtenez les prix de vente les plus récents

    Paramètres Est-ce que cela est nécessaire? Décrire
    prix actuel Oui, oui. Envoyez le prix pour trouver le prix de vente le plus récent.
  • base_position (titre)

    Le sous-sol

    Paramètres Est-ce que cela est nécessaire? Décrire
    le ticker Oui, oui. Ouvrir le boîtier pour ouvrir la grille, cette fonction exécute la fonction de rappeleventLes événementsbase_position
  • ajouter_ordre ((ordre_id)

    Ajoutez des listes à la grille

    Paramètres Est-ce que cela est nécessaire? Décrire
    Nom de l'équipe Oui, oui. La fonction id pour ajouter un lien vers le haut de la grille, pour envoyer des sous-dossiers ou des ordres de transaction va trouver la grille vers le bas de cette id et exécuter la fonction de rappel.eventLes événementsadd_order
  • annuler la commande

    Retraite des commandes

    Paramètres Est-ce que cela est nécessaire? Décrire
    Nom de l'équipe Oui, oui. Révoquer une commande spécifiée qui exécute une fonction de rappeleventLes événementscancel_order

Les événements

Les événements sont des fonctions de rappel spécifiées qui sont appelées lors de l'exécution d'une fonction.

gm = GridPriceManager(1000, 800, 10)

# 底仓事件,在调用base_position方法时会触发此事件
@gm.event('base_position')
def base_position(price):
    # 传入最近的网格价格,以此价格作为买入价格参考
    print(price)
    return 123456	# 返回底仓订单,manger将订单记录
Les événements Est-ce que cela est nécessaire? La transmission Retour
base_position Oui, oui. prix, prix d'achat, type de flotteur Id de la commande au sous-sol
ajouter_ordre - Je ne sais pas. prix, prix d'achat sur la grille, type de dict, {dict up: prix sur la grille, dict down: prix en bas de la grille} Avec le dict du même format, correspondant à l'id de transaction de la grille supérieure et à l'id de transaction de la grille inférieure
annuler la commande - Je ne sais pas. order_id, spécifier le type d'ordre id, int ou str à annuler bool, réussir à annuler
changement - Je ne sais pas. liste de grille L'incident a été déclenché par un changement d'information dans la grille.

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)

Plus de