| 매개 변수 | 필요한가요? | 설명 |
|---|---|---|
| upper_price | 아니 | 인터넷 상의 경계 가격 |
| 더 낮은 가격 | 아니 | 그레이트 아래의 경계 가격 |
| grid_num | 아니 | 그레이트 수 (비등) |
| 간격 | 아니 | 격자 간격 (等比) |
| 옆면 | 아니 | 지원 전송long，short이 글은 필자가 작성한 글입니다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 | 네 | 네트워크 가격과 주문 정보는 키가 값으로 저장되며, 키는 가격이며, 값은 주문 id입니다. |
| 간격 | 네 | |
| upper_price | 네 | |
| 더 낮은 가격 | 네 | |
| grid_num | 네 | |
| 옆면 | 네 | |
| grid_diff | 네 | |
| 종류 | 네 | 0과 같고 |
get_nearest_buy_price (현재의 가격)
최근 매출 가격을 확인합니다.
| 매개 변수 | 필요한가요? | 설명 |
|---|---|---|
| 현재 가격 | 네 | 입력값을 통해 가장 최근 구매 가격을 찾습니다. |
get_nearest_sell_price (현재의 가격)
최근 매출액을 확인합니다.
| 매개 변수 | 필요한가요? | 설명 |
|---|---|---|
| 현재 가격 | 네 | 입력값을 통해 가장 최근 판매값을 찾습니다. |
base_position (티커)
기저구
| 매개 변수 | 필요한가요? | 설명 |
|---|---|---|
| 틱 | 네 | 풀 하우스를 열고 그레이스를 열고, 이 함수는 호출 함수를 실행합니다.event사건base_position |
add_order (order_id)
| 매개 변수 | 필요한가요? | 설명 |
|---|---|---|
| order_id | 네 | 격자 상하에 기열된 문서를 추가하거나 하위 저장소 또는 거래 주문을 전송하는 id 함수는 이 id의 상하 격자를 찾습니다. 이 함수는 호출 함수를 실행합니다.event사건add_order |
cancel_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将订单记录
| 사건 | 필요한가요? | 침입 | 돌아갑니다. |
|---|---|---|---|
| base_position | 네 | 가격, 구매 가격, 플로트 타입 | 하위 보관소 주문 id |
| add_order | 아니 | 같은 형식으로 전달되는 dikte와 대응하는 위 격자 트랜잭션 id, 아래 격자 트랜잭션 id | |
| cancel_order 를 취소합니다 | 아니 | order_id, 취소되는 order 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)