forked from kodorvan/stcs
130 lines
4.2 KiB
Python
130 lines
4.2 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("get_positions")
|
|
|
|
|
|
async def get_active_positions(tg_id: int) -> list | None:
|
|
"""
|
|
Get active positions for a user
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
response = client.get_positions(category="linear", settleCoin="USDT")
|
|
|
|
if response["retCode"] == 0:
|
|
positions = response.get("result", {}).get("list", [])
|
|
active_symbols = [
|
|
pos.get("symbol") for pos in positions if float(pos.get("size", 0)) > 0
|
|
]
|
|
if active_symbols:
|
|
logger.info("Active positions for user: %s", tg_id)
|
|
return positions
|
|
else:
|
|
logger.warning("No active positions found for user: %s", tg_id)
|
|
return ["No active positions found"]
|
|
else:
|
|
logger.error(
|
|
"Error getting active positions for user %s: %s",
|
|
tg_id,
|
|
response["retMsg"],
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.error("Error getting active positions for user %s: %s", tg_id, e)
|
|
return None
|
|
|
|
|
|
async def get_active_positions_by_symbol(tg_id: int, symbol: str) -> dict | None:
|
|
"""
|
|
Get active positions for a user by symbol
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
response = client.get_positions(category="linear", symbol=symbol)
|
|
|
|
if response["retCode"] == 0:
|
|
positions = response.get("result", {}).get("list", [])
|
|
if positions:
|
|
logger.info("Active positions for user: %s", tg_id)
|
|
return positions
|
|
else:
|
|
logger.warning("No active positions found for user: %s", tg_id)
|
|
return None
|
|
else:
|
|
logger.error(
|
|
"Error getting active positions for user %s: %s",
|
|
tg_id,
|
|
response["retMsg"],
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.error("Error getting active positions for user %s: %s", tg_id, e)
|
|
return None
|
|
|
|
|
|
async def get_active_orders(tg_id: int) -> list | None:
|
|
"""
|
|
Get active orders
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
response = client.get_open_orders(
|
|
category="linear",
|
|
settleCoin="USDT",
|
|
limit=50,
|
|
)
|
|
|
|
if response["retCode"] == 0:
|
|
orders = response.get("result", {}).get("list", [])
|
|
active_orders = [
|
|
pos.get("symbol") for pos in orders if float(pos.get("qty", 0)) > 0
|
|
]
|
|
if active_orders:
|
|
logger.info("Active orders for user: %s", tg_id)
|
|
return orders
|
|
else:
|
|
logger.warning("No active orders found for user: %s", tg_id)
|
|
return ["No active orders found"]
|
|
else:
|
|
logger.error(
|
|
"Error getting active orders for user %s: %s", tg_id, response["retMsg"]
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.error("Error getting active orders for user %s: %s", tg_id, e)
|
|
return None
|
|
|
|
|
|
async def get_active_orders_by_symbol(tg_id: int, symbol: str) -> dict | None:
|
|
"""
|
|
Get active orders by symbol
|
|
"""
|
|
try:
|
|
client = await get_bybit_client(tg_id)
|
|
response = client.get_open_orders(
|
|
category="linear",
|
|
symbol=symbol,
|
|
limit=50,
|
|
)
|
|
|
|
if response["retCode"] == 0:
|
|
orders = response.get("result", {}).get("list", [])
|
|
if orders:
|
|
logger.info("Active orders for user: %s", tg_id)
|
|
return orders
|
|
else:
|
|
logger.warning("No active orders found for user: %s", tg_id)
|
|
return None
|
|
else:
|
|
logger.error(
|
|
"Error getting active orders for user %s: %s", tg_id, response["retMsg"]
|
|
)
|
|
return None
|
|
except Exception as e:
|
|
logger.error("Error getting active orders for user %s: %s", tg_id, e)
|
|
return None
|