from decimal import Decimal
from t_tech.invest import OrderDirection, OrderType
from t_tech.invest.utils import quotation_to_decimal as to_dec
import uuid
class RebalanceBot:
def __init__(
self,
client,
account_id: str,
target_weights: dict,
corridor: Decimal,
dry_run=True,
):
self.client = client
self.account_id = account_id
self.target_weights = target_weights
self.corridor = corridor
self.portfolio_value = Decimal("0")
self.instrument_cache = {}
self.dry_run = dry_run
def fetch_portfolio(self):
account = self.client.operations.get_portfolio(account_id=self.account_id)
self.portfolio_value = to_dec(account.total_amount_portfolio)
return account.positions
def get_instrument_data(self, instrument_uid):
if instrument_uid not in self.instrument_cache:
resp = self.client.instruments.find_instrument(query=instrument_uid)
self.instrument_cache[instrument_uid] = resp.instruments[0]
return self.instrument_cache[instrument_uid]
def calculate_rebalance(self, positions, target_config) -> tuple[list[dict], str]:
plan = []
plan_report = "\n📊 План ребаланса\n"
allowed_uids = {info["uid"] for info in target_config.values()}
current_values = {}
current_prices = {}
for pos in positions:
# Пропускаем кэш (рубли) только при проверке лишних активов,
# т.к. кэш учитывается в target_config как RUB
if pos.instrument_type == "currency":
continue
if pos.instrument_uid not in allowed_uids:
ticker = pos.ticker
qty = to_dec(pos.quantity)
if qty > 0:
# Получаем данные об инструменте для определения лотности
instr_info = self.get_instrument_data(pos.instrument_uid)
# В Тинькофф продажа идет в лотах
lots_to_sell = int(qty / instr_info.lot)
if lots_to_sell > 0:
print(
f"🗑️ ЛИШНИЙ АКТИВ: {ticker} (нет в конфиге). Продаем {lots_to_sell} лотов."
)
plan_report += f"• SELL {ticker}: {lots_to_sell} лотов (не в конфиге)\n"
plan.append(
{
"ticker": ticker,
"figi": pos.figi,
"action": "SELL",
"lots": lots_to_sell,
"reason": "not_in_config",
}
)
current_map = {pos.instrument_uid: pos for pos in positions}
for ticker, info in target_config.items():
uid = info["uid"]
target_weight = info["weight"]
pos = current_map.get(uid)
if pos is None:
current_weight = Decimal("0")
current_value = Decimal("0")
else:
price = to_dec(pos.current_price)
qty = to_dec(pos.quantity)
current_value = price * qty
current_weight = current_value / self.portfolio_value
if ticker != "RUB":
current_prices[ticker] = price
current_values[ticker] = current_value
delta = target_weight - current_weight
money_delta = delta * self.portfolio_value
print(
f"{ticker}, Текущая доля {current_weight:.2%}, Цель {target_weight:.2%}, Дельта: {delta:.2%}, Дельта в рублях: {money_delta:.6}"
)
if abs(delta) > self.corridor:
# Для RUB не создаём ордера, только выводим информацию
if ticker == "RUB":
if delta > 0:
plan_report += f"• ⚠️ Недостаток кэша: нужно {abs(delta * self.portfolio_value):.2f} руб. (уменьшите покупки)\n"
else:
plan_report += f"• 💰 Избыток кэша: {abs(delta * self.portfolio_value):.2f} руб. доступно для инвестиций\n"
else:
money_to_trade = self.portfolio_value * delta
instrument_info = self.get_instrument_data(uid)
one_lot_price = price * instrument_info.lot
lots = int(money_to_trade / one_lot_price)
if lots != 0:
plan.append(
{
"figi": instrument_info.figi,
"ticker": ticker,
"uid": uid,
"action": "BUY" if lots > 0 else "SELL",
"lots": abs(lots),
"delta_pct": delta * 100,
}
)
plan_report += f"• {'BUY' if lots > 0 else 'SELL'} {ticker}: {abs(lots)} лотов; на сумму {money_to_trade:.2f} руб.\n"
print(
f"{ticker:12} | Доля: {current_weight:6.2%} | Цель: {target_weight:6.2%}"
)
# Инвестируем избыток кэша, даже если остальные отклонения в пределах коридора
if "RUB" in target_config and "RUB" in current_values:
target_cash = self.portfolio_value * target_config["RUB"]["weight"]
cash_excess = current_values["RUB"] - target_cash
if cash_excess > 0:
plan_report += f"• 💰 Избыток кэша: {cash_excess:.2f} руб. будет распределён по целям\n"
non_rub_weights = {
t: info["weight"]
for t, info in target_config.items()
if t != "RUB"
}
total_non_rub_weight = sum(non_rub_weights.values())
if total_non_rub_weight > 0:
for ticker, weight in non_rub_weights.items():
uid = target_config[ticker]["uid"]
share = weight / total_non_rub_weight
money_to_trade = cash_excess * share
price = current_prices.get(ticker)
if price is None or price <= 0:
plan_report += f"• ⚠️ Пропуск {ticker}: нет цены для расчёта лотов\n"
continue
instrument_info = self.get_instrument_data(uid)
one_lot_price = price * instrument_info.lot
lots = int(money_to_trade / one_lot_price)
if lots > 0:
plan.append(
{
"figi": instrument_info.figi,
"ticker": ticker,
"uid": uid,
"action": "BUY",
"lots": lots,
"delta_pct": Decimal("0"),
"reason": "cash_excess",
}
)
plan_report += f"• BUY {ticker}: {lots} лотов; на сумму {money_to_trade:.2f} руб. (излишек кэша)\n"
plan_report += "\n— Конец плана ребаланса —\n"
return plan, plan_report
def execute_orders(self, trades, group_name) -> str:
execute_report = f"\n⚙️ Исполнение: {group_name}\n"
if not trades:
execute_report += "Нет сделок для исполнения.\n"
return execute_report
print(f"\n--- Исполнение блока: {group_name} ---")
for trade in trades:
action_ru = "КУПИТЬ" if trade["action"] == "BUY" else "ПРОДАТЬ"
if self.dry_run:
execute_report += f"• Симуляция: {action_ru} {trade['ticker']} — {trade['lots']} лотов\n"
print(
f"[СИМУЛЯЦИЯ] {action_ru} {trade['ticker']}: {trade['lots']} лотов"
)
continue
try:
direction = (
OrderDirection.ORDER_DIRECTION_BUY
if trade["action"] == "BUY"
else OrderDirection.ORDER_DIRECTION_SELL
)
response = self.client.orders.post_order(
figi=trade["figi"],
quantity=int(trade["lots"]),
direction=direction,
account_id=self.account_id,
order_type=OrderType.ORDER_TYPE_MARKET,
order_id=str(uuid.uuid4()),
)
execute_report += f"• ✅ {action_ru} {trade['ticker']} — {trade['lots']} лотов. ID: {response.order_id}\n"
print(
f"[ИСПОЛНЕНО]\n{trade['ticker']} на {trade['lots']} лотов. ID: {response.order_id}"
)
except Exception as e:
execute_report += f"• ❌ Ошибка по {trade['ticker']}: {e}\n"
print(f"[ОШИБКА]\nПри сделке с {trade['ticker']}: {e}")
return execute_report