格子管理ツール

作者: リン・ハーンプログラム, 日付: 2023-04-30 09:03:58
タグ:

パラメータ 必要なのか 記述
upper_price (上値) はい ネットワーク上の境界価格
低価格 はい 格子下限価格
grid_num はい 格子数 (等差)
インターバル はい 格子間隔 (等比)
はい 支援配信longshort記入しないのはデフォルトです.long
データ はい 格子情報,字典のタイプがあります
# 创建了一个价格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)

データ構造

パラメータ 必要なのか 記述
grid_list について そうだ 格子価格と注文情報は,キー値で保存され,keyは価格,valueは注文id
インターバル そうだ
upper_price (上値) そうだ
低価格 そうだ
grid_num そうだ
そうだ
grid_diff そうだ
タイプ そうだ 負の2倍に等しいです.

関数

  • get_nearest_buy_price (現在の価格)

    最近の格子購入価格を表示する

    パラメータ 必要なのか 記述
    現行価格 そうだ 価格を入力して,最近購入した価格を見つけます.
  • get_nearest_sell_price (現在の価格)

    最近のネット販売価格を入手

    パラメータ 必要なのか 記述
    現行価格 そうだ 価格を入力して,最近売れた価格を見つけます.
  • バース_ポジション (ティッカー)

    底棚

    パラメータ 必要なのか 記述
    ティッカー そうだ 格子を開くため,この関数は呼び戻し関数を実行します.event事件base_position
  • add_order ((order_id) を追加する

    格子上のリストを追加

    パラメータ 必要なのか 記述
    order_id そうだ 格子上に下列列列を追加し,底辺に転送するid関数または取引注文は,このidの上下列格子を見つけ,この関数は呼び戻し関数を実行します.event事件add_order
  • キャンセル_オーダー ((order_id)

    注文を取り消す

    パラメータ 必要なのか 記述
    order_id そうだ 指定された注文をキャンセルします. この関数は呼び戻し関数を実行します.event事件cancel_order

事件

イベントとは,関数の実行中に呼び出される指定呼び出し関数である.ここで,常に event関数を使用して指定されたイベント,デコラターモードに伝達する.

gm = GridPriceManager(1000, 800, 10)

# 底仓事件,在调用base_position方法时会触发此事件
@gm.event('base_position')
def base_position(price):
    # 传入最近的网格价格,以此价格作为买入价格参考
    print(price)
    return 123456	# 返回底仓订单,manger将订单记录
事件 必要なのか 感染 戻る
ベース_ポジション そうだ 価格,購入価格,フロートタイプ 倉庫の注文 id
add_order を追加する はい price,入網価格,dictタイプ,{up:上網価格,down:下網価格} 同形式の伝達されたdictに対応する上格交换id,下格交换id
キャンセル_オーダー はい order_id,キャンセルされる注文id,int,またはstrのタイプを指定する bool,キャンセル成功か
変化 はい 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)

もっと