forked from kodorvan/stcs
101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
import logging.config
|
|
|
|
from app.bybit import get_bybit_client
|
|
from app.bybit.logger_bybit.logger_bybit import LOGGING_CONFIG
|
|
|
|
logging.config.dictConfig(LOGGING_CONFIG)
|
|
logger = logging.getLogger("close_positions")
|
|
|
|
|
|
async def close_position_by_symbol(
|
|
tg_id: int, symbol: str
|
|
) -> bool:
|
|
"""
|
|
Closes all positions
|
|
:param tg_id: Telegram user ID
|
|
:param symbol: symbol
|
|
:return: bool
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
|
|
response = client.get_positions(
|
|
category="linear", symbol=symbol
|
|
)
|
|
positions = response.get("result", {}).get("list", [])
|
|
r_side = "Sell" if positions[0].get("side") == "Buy" else "Buy"
|
|
qty = positions[0].get("size")
|
|
position_idx = positions[0].get("positionIdx")
|
|
|
|
response = client.place_order(
|
|
category="linear",
|
|
symbol=symbol,
|
|
side=r_side,
|
|
orderType="Market",
|
|
qty=qty,
|
|
timeInForce="GTC",
|
|
positionIdx=position_idx,
|
|
)
|
|
if response["retCode"] == 0:
|
|
logger.info("Positions closed for %s for user %s", symbol, tg_id)
|
|
return True
|
|
else:
|
|
logger.error(
|
|
"Error closing position for %s for user %s", symbol, tg_id
|
|
)
|
|
return False
|
|
except Exception as e:
|
|
logger.error(
|
|
"Error closing positions for %s for user %s: %s", symbol, tg_id, e
|
|
)
|
|
return False
|
|
|
|
|
|
async def cancel_order(tg_id: int, symbol: str, order_id: str) -> bool:
|
|
"""
|
|
Cancel order by order id
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
|
|
cancel_resp = client.cancel_order(
|
|
category="linear", symbol=symbol, orderId=order_id
|
|
)
|
|
|
|
if cancel_resp.get("retCode") == 0:
|
|
return True
|
|
else:
|
|
logger.error(
|
|
"Error canceling order for user %s: %s",
|
|
tg_id,
|
|
cancel_resp.get("retMsg"),
|
|
)
|
|
return False
|
|
except Exception as e:
|
|
logger.error("Error canceling order for user %s: %s", tg_id, e)
|
|
return False
|
|
|
|
|
|
async def cancel_all_orders(tg_id: int) -> bool:
|
|
"""
|
|
Cancel all open orders
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
cancel_resp = client.cancel_all_orders(category="linear", settleCoin="USDT")
|
|
|
|
if cancel_resp.get("retCode") == 0:
|
|
logger.info("All orders canceled for user %s", tg_id)
|
|
return True
|
|
else:
|
|
logger.error(
|
|
"Error canceling order for user %s: %s",
|
|
tg_id,
|
|
cancel_resp.get("retMsg"),
|
|
)
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error("Error canceling order for user %s: %s", tg_id, e)
|
|
return False
|