kodorvan/stcs
7
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed close position

Browse Source
This commit is contained in:
algizn97
2025-10-03 14:18:53 +05:00
parent 6705bf4492
commit 4adbd70948
2 changed files with 51 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
app/bybit/close_positions.py
View File

@@ -1,77 +1,69 @@
import logging.config
from app.bybit import get_bybit_client
from app.bybit.get_functions.get_positions import get_active_positions_by_symbol
from app.bybit.logger_bybit.logger_bybit import LOGGING_CONFIG
logging.config.dictConfig(LOGGING_CONFIG)
logger = logging.getLogger("close_positions")
async def close_position(tg_id: int, symbol: str) -> bool:
async def close_position(tg_id: int, symbol: str, side: str, position_idx: int, qty: float) -> bool:
"""
Closes position
Closes all positions
:param tg_id: Telegram user ID
:param symbol: symbol
:param side: side
:param position_idx: position index
:param qty: quantity
:return: bool
"""
try:
client = await get_bybit_client(tg_id)
active_positions = await get_active_positions_by_symbol(tg_id, symbol)
side = active_positions.get("side")
size = active_positions.get("size")
position_idx = active_positions.get("positionIdx")
if side == "Buy":
side = "Sell"
elif side == "Sell":
side = "Buy"
r_side = "Sell"
else:
r_side = "Buy"
response = client.place_order(
category="linear",
symbol=symbol,
side=side,
side=r_side,
orderType="Market",
qty=size,
qty=qty,
timeInForce="GTC",
positionIdx=position_idx,
)
if response["retCode"] == 0:
logger.info("Position closed for %s for user %s", symbol, tg_id)
logger.info("All 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)
logger.error("Error closing all positions for %s for user %s", symbol, tg_id)
return False
except Exception as e:
logger.error("Error closing position for %s for user %s: %s", symbol, tg_id, e)
logger.error("Error closing all positions for %s for user %s: %s", symbol, tg_id, e)
return False
async def cancel_order(tg_id, symbol) -> bool:
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)
orders_resp = client.get_open_orders(category="linear", symbol=symbol)
orders = orders_resp.get("result", {}).get("list", [])
for order in orders:
order_id = order.get("orderId")
cancel_resp = client.cancel_order(
category="linear", symbol=symbol, orderId=order_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"),
)
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
return False
return False
except Exception as e:
logger.error("Error canceling order for user %s: %s", tg_id, e)
return False