import asyncio
import logging
import sys
import os
import json
import subprocess
import signal
import psutil
import time
import re
import pika
import pandas as pd
import threading
from openpyxl.styles import Font
from datetime import datetime, timezone, timedelta
from pathlib import Path
from aiogram.types import Message
from aiogram.enums import ParseMode
from aiogram import Bot, Dispatcher, types
from aiogram.filters import CommandObject
from aiogram.filters.command import Command
from io import BytesIO
from aiogram.types import FSInputFile
from aiogram import F
from aiogram.exceptions import TelegramBadRequest
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from db_fetcher import DBReader
from manual import manual_get_ubook2_async

logging.basicConfig(level=logging.INFO)
debug = False

##### RABBIT #######
depth = {}

def rabbit_consumer():
    def callback(ch, method, properties, body):
        ch.basic_ack(delivery_tag=method.delivery_tag)

        msg = json.loads(body.decode())
        name = '_'.join(msg['name'].split('.')[:-1])

        if name not in depth:
            depth[name] = {'ts': 0, 'depth': {'asks': [], 'bids': []}}

        if msg['mtype'] != 'total_depth':
            print(f"[warning] {int(1000*time.time())} not total_depth type: {msg['mtype']}")
            return

        if int(msg['ts']) < depth[name]['ts']:
            print(f"[debug] {int(1000*time.time())} old message: {int(msg['ts'])} < {depth[name]['ts']}")
            return

        depth[name] = {'ts': int(msg['ts']), 'depth': msg['msg']}

    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='localhost', port=9669)
    )
    channel = connection.channel()
    channel.queue_declare(queue='depth_queue.prod', durable=True, arguments={'x-message-ttl': 3600000, 'x-queue-mode': 'lazy'})
    channel.basic_consume(queue='depth_queue.prod', on_message_callback=callback)
    channel.start_consuming()

rabbit_thread = threading.Thread(target=rabbit_consumer, daemon=True)
rabbit_thread.start()
####################

scheduler = AsyncIOScheduler()
dbReader = DBReader('/home/ubuntu/app/database.ini')
dbSideCashoutReader = DBReader('/home/ubuntu/app/side_tb_database.ini')
bot = Bot(token="8196473785:AAFCkXVPa7X0niky7RTHjjfA_-jMXodyWY0")
dp = Dispatcher()

def parse_clients_json():
    with open('clients.json', 'r') as clients_f:
        chat_ids = json.loads(''.join(clients_f.readlines()).replace(' ', '').replace('\n', ''))
        return chat_ids

def parse_admins_json():
    with open('admins.json', 'r') as admins_f:
        admin_ids = json.loads(''.join(admins_f.readlines()).replace(' ', '').replace('\n', ''))
        return admin_ids

def parse_alarms_json():
    with open('/home/ubuntu/tgbot/chat-id-token.json', 'r') as admins_f:
        # chats_mapping: по токену получить айди чата
        chats_mapping = json.loads(''.join(admins_f.readlines()).replace(' ', '').replace('\n', ''))
    return chats_mapping

chats_mapping = parse_alarms_json()
chat_ids = parse_clients_json()
admin_ids = parse_admins_json()

timezones = {
    -4788418340: timedelta(hours=9),
    # -1002150192437: timedelta(hours=3),
}

@dp.message(Command("liquidity_volume"))
async def volume_balances(message: types.Message):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    result = ''
    await dbReader.init_pool()
    tasks = [dbReader.get_volumes_by_group(group) for group in chat_ids[chat_id]]
    result = await asyncio.gather(*tasks, return_exceptions=True)
    await message.answer(''.join([res for res in result if res is not None and not isinstance(res, Exception)]))


MAX_LENGTH = 4096


def split_html_safe(text, max_length=MAX_LENGTH):
    """Split text into chunks <= max_length, breaking only on blank-line
    boundaries between symbol blocks so HTML tags stay intact."""
    blocks = text.split('\n\n')
    chunks = []
    buf = ''
    for block in blocks:
        piece = block + '\n\n'
        if len(buf) + len(piece) > max_length:
            if buf:
                chunks.append(buf)
                buf = ''
            # block itself bigger than limit — fall back to line-level split
            if len(piece) > max_length:
                for line in piece.split('\n'):
                    line += '\n'
                    if len(buf) + len(line) > max_length and buf:
                        chunks.append(buf)
                        buf = ''
                    buf += line
                continue
        buf += piece
    if buf:
        chunks.append(buf)
    return chunks


@dp.message(Command("trading_balances"))
async def trading_balances(message: types.Message):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    result = ''
    await dbReader.init_pool()
    tasks = [dbReader.get_balances_by_group(group, 'V') for group in chat_ids[chat_id]]
    result = await asyncio.gather(*tasks, return_exceptions=True)
    text = ''.join([res for res in result if res is not None and not isinstance(res, Exception)])
    for i in range(0, len(text), MAX_LENGTH):
        await message.answer(text[i:i+MAX_LENGTH])


@dp.message(Command("mm_balances"))
async def mm_balance(message: types.Message):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    result = ''
    await dbReader.init_pool()
    tasks = [dbReader.get_balances_by_group(group, 'M') for group in chat_ids[chat_id]]
    result = await asyncio.gather(*tasks, return_exceptions=True)
    text = ''.join([res for res in result if res is not None and not isinstance(res, Exception)])
    for i in range(0, len(text), MAX_LENGTH):
        await message.answer(text[i:i+MAX_LENGTH])

def parse_date_range(args_text):
    """Parse date range from command arguments.
    Supported formats: DD.MM or DD.MM.YYYY
    Returns (date_from, date_to) as strings 'YYYY-MM-DD', or (None, None) on failure.
    Without arguments returns last 7 days.
    """
    if not args_text or not args_text.strip():
        today = datetime.now().date()
        date_from = today - timedelta(days=7)
        return str(date_from), str(today)

    parts = args_text.strip().split()
    if len(parts) != 2:
        return None, None

    dates = []
    for p in parts:
        segments = p.split('.')
        try:
            if len(segments) == 2:
                d = datetime.strptime(p, '%d.%m').replace(year=datetime.now().year).date()
            elif len(segments) == 3:
                d = datetime.strptime(p, '%d.%m.%Y').date()
            else:
                return None, None
            dates.append(d)
        except ValueError:
            return None, None

    return str(dates[0]), str(dates[1])


@dp.message(Command("report"))
async def report(message: types.Message, command: CommandObject):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    date_from, date_to = parse_date_range(command.args)
    if date_from is None:
        await message.answer("Usage: /report DD.MM DD.MM\nExample: /report 04.04 14.04")
        return

    await dbReader.init_pool()
    await dbSideCashoutReader.init_pool(side=True)
    # Symbols list is cached at first init_pool; refresh so newly added
    # side instances (e.g. tokens added in side DB after bot start) are seen.
    await dbReader.update_symbols_list()
    await dbSideCashoutReader.update_symbols_list(side=True)
    tasks = [
        dbReader.get_volume_report_by_group(group, date_from, date_to)
        for group in chat_ids[chat_id] if '.' not in group
    ]
    tasks.extend([
        dbSideCashoutReader.get_side_cashout_report_by_group(group, date_from, date_to)
        for group in chat_ids[chat_id] if '.' in group
    ])
    result = await asyncio.gather(*tasks, return_exceptions=True)
    text = ''.join([res for res in result if res is not None and not isinstance(res, Exception)])
    if not text:
        await message.answer("No data for report")
        return
    for chunk in split_html_safe(text):
        await message.answer(chunk, parse_mode=ParseMode.HTML)


@dp.message(Command("mm_report"))
async def mm_report(message: types.Message, command: CommandObject):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    date_from, date_to = parse_date_range(command.args)
    if date_from is None:
        await message.answer("Usage: /mm_report DD.MM DD.MM\nExample: /mm_report 04.04 14.04")
        return

    await dbReader.init_pool()
    await dbReader.update_symbols_list()
    tasks = [dbReader.get_mm_report_by_group(group, date_from, date_to) for group in chat_ids[chat_id]]
    result = await asyncio.gather(*tasks, return_exceptions=True)
    text = ''.join([res for res in result if res is not None and not isinstance(res, Exception)])
    if not text:
        await message.answer("No data for MM report")
        return
    for chunk in split_html_safe(text):
        await message.answer(chunk, parse_mode=ParseMode.HTML)


@dp.message(Command("ping"))
async def ping(message: types.Message):
    await message.reply("pong")

@dp.message(Command("cashout_balances"))
async def cashout_balances(message: types.Message):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    result = ''
    await dbReader.init_pool()
    await dbSideCashoutReader.init_pool(side=True)
    tasks_tb = [dbReader.get_balances_by_group(group, 'T') for group in chat_ids[chat_id] if '.' not in group]
    tasks_tb.extend([dbSideCashoutReader.get_balances_by_group(group) for group in chat_ids[chat_id] if '.' in group])

    result = await asyncio.gather(*tasks_tb, return_exceptions=True)
    await message.answer(''.join([res for res in result if res is not None and not isinstance(res, Exception)]))


def calculate_book_orders(symbol): # NOTE: проверка на сущесвование symbol делается перед вызовом calculate_book_orders
    book = depth[symbol]['depth']
    book['asks'] = sorted(book['asks'], key=lambda x: x[0], reverse=True)
    book['bids'] = sorted(book['bids'], key=lambda x: x[0], reverse=True)

    best_ask = min(book['asks'], key=lambda x: float(x[0]))[0]
    best_bid = max(book['bids'], key=lambda x: float(x[0]))[0]
    book_date = datetime.fromtimestamp(depth[symbol]['ts'] / 1000).strftime('%Y-%m-%d %H:%M:%S')

    orders = []
    cur_info_asks = {2: [0, 0], 5: [0, 0], 10: [0, 0], 15: [0, 0], 20: [0, 0], 30: [0, 0], 50: [0, 0], 100: [0, 0]}
    cur_info_bids = {2: [0, 0], 5: [0, 0], 10: [0, 0], 15: [0, 0], 20: [0, 0], 30: [0, 0], 50: [0, 0], 100: [0, 0]}

    for order in book['asks']:
        prct = 100 * ((float(order[0]) / best_ask) - 1)
        for key in cur_info_asks:
            if prct <= key:
                cur_info_asks[key][0] += float(order[1])
                cur_info_asks[key][1] += float(order[0])*float(order[1])

    for order in book['asks'][-100:]:
        orders.append({'symbol': symbol, 'time': book_date, 'price': order[0], 'quantity': order[1], 'side': 'SELL'})

    for order in book['bids']:
        prct = 100 * ((best_bid / float(order[0])) - 1)
        for key in cur_info_bids:
            if prct <= key:
                cur_info_bids[key][0] += float(order[1])
                cur_info_bids[key][1] += float(order[0])*float(order[1])

    for order in book['bids'][:100]:
        orders.append({'symbol': symbol, 'time': book_date, 'price': order[0], 'quantity': order[1], 'side': 'BUY'})

    orders.append({'symbol': None, 'time': None, 'price': None, 'side': None})

    return [cur_info_bids, cur_info_asks], orders


@dp.message(Command(re.compile(r"depth(_\w+)?")))
async def get_depth(message: types.Message, command: CommandObject):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    exchange = None
    if command.args:
        exchange = command.args.strip().lower()
    else:
        match = re.match(r"/depth_(\w+)", message.text)
        if match:
            exchange = match.group(1).lower()
    if exchange == 'all':
        exchange = None

    total_info = {}
    for group in chat_ids[chat_id]:
        symbol = group + '_USDT_' + exchange
        if symbol not in depth:
            continue
        total_info[symbol], _ = calculate_book_orders(symbol)

    result = []
    for symbol, depth_info in total_info.items():
        symbol_orders = []
        for prct, values in depth_info[0].items():
            symbol_orders.append({'symbol': symbol, 'percent': prct, 'cumulative quantity': values[0], 'cumulative amount': values[1], 'side': 'BUY'})
        for prct, values in depth_info[1].items():
            symbol_orders.append({'symbol': symbol, 'percent': prct, 'cumulative quantity': values[0], 'cumulative amount': values[1], 'side': 'SELL'})
        symbol_orders.append({'symbol': None, 'percent': None, 'cumulative quantity': None, 'cumulative amount': None, 'side': None})

        symbol_orders = pd.DataFrame(symbol_orders)
        sell_sorted = symbol_orders[symbol_orders['side'] == 'SELL'].sort_values('percent', ascending=False)
        buy_sorted = symbol_orders[symbol_orders['side'] == 'BUY'].sort_values('percent', ascending=True)
        result.append(pd.concat([sell_sorted, buy_sorted], ignore_index=True))

    result = pd.concat(result, ignore_index=True)
    output = BytesIO()
    with pd.ExcelWriter(output, engine='openpyxl') as writer:
        result.to_excel(writer, index=False, sheet_name='Depth %')

        worksheet = writer.sheets['Depth %']
        for column_cells in worksheet.columns:
            length = max(len(str(cell.value)) for cell in column_cells)
            worksheet.column_dimensions[column_cells[0].column_letter].width = length + 2

    output.seek(0)

    await message.answer_document(
        types.BufferedInputFile(
            file=output.read(),
            filename=f"depth_{exchange if exchange else 'all'}.xlsx"
        ),
        caption=f"Excel file with depth{' for ' + exchange if exchange else ''}"
    )

@dp.message(Command(re.compile(r"orderbook(_\w+)?")))
async def orders(message: types.Message, command: CommandObject):
    chat_id = str(message.chat.id)
    if chat_id not in chat_ids:
        await message.answer("Your chat is not in whitelist")
        return

    exchange = None
    if command.args:
        exchange = command.args.strip().lower()
    else:
        match = re.match(r"/orderbook_(\w+)", message.text)
        if match:
            exchange = match.group(1).lower()
    if exchange == 'all':
        exchange = None

    await dbReader.init_pool()
    tasks = [dbReader.get_active_orders(group) for group in chat_ids[chat_id]]
    result = await asyncio.gather(*tasks, return_exceptions=True)

    result_as_list = []
    for res in result:
        if res is not None and not isinstance(res, Exception):
            result_as_list.extend(res)

    result = [pd.DataFrame(res) for res in result if res is not None and not isinstance(res, Exception)]

    total_orders = []
    for group in chat_ids[chat_id]:
        symbol = group + '_USDT_' + exchange
        if symbol not in depth:
            continue
        _, orders = calculate_book_orders(symbol)
        total_orders.extend(orders)

    markups = set()
    for own_order in result_as_list:
        for depth_order in total_orders:
            if own_order['symbol'].replace('/', '_').replace(' ', '_') == depth_order['symbol'] and own_order['price'] == depth_order['price']:
                markups.add(f"{depth_order['symbol']}/{depth_order['price']}")

    all_orders_df = pd.concat(result, ignore_index=True)
    if exchange is not None:
        all_orders_df = all_orders_df[all_orders_df['symbol'].str.contains(exchange, case=False, na=False)]
    all_orders_df = all_orders_df.reset_index(drop=True)

    output = BytesIO()
    with pd.ExcelWriter(output, engine='openpyxl') as writer:
        all_orders_df = all_orders_df.copy()
        all_orders_df['side_order'] = all_orders_df['side'].map({'SELL': 0, 'BUY': 1})
        all_orders_df['price_order'] = all_orders_df.apply(
            lambda x: -x['price'] if x['side']=='SELL' else x['price'], axis=1
        )
        all_orders_df = all_orders_df.sort_values(
            by=['symbol', 'side_order', 'price_order'],
            ascending=[True, True, True]
        )
        all_orders_df = all_orders_df.drop(columns=['side_order', 'price_order'])
        all_orders_df = all_orders_df.reset_index()

        if not all_orders_df.empty:
            all_orders_df.to_excel(writer, index=True, sheet_name='Actual Orders')

        depth_df = pd.DataFrame(total_orders)
        if not depth_df.empty:
            depth_df.to_excel(writer, index=True, sheet_name='Depth')

        for sheet_name in writer.sheets:
            worksheet = writer.sheets[sheet_name]
            for column_cells in worksheet.columns:
                length = max(len(str(cell.value)) for cell in column_cells)
                worksheet.column_dimensions[column_cells[0].column_letter].width = length + 2

        if 'Depth' in writer.sheets:
            depth_ws = writer.sheets['Depth']
            for idx, row in depth_df.iterrows():
                key = f"{row['symbol']}/{row['price']}"
                if key in markups:
                    excel_row = idx + 2  # 1 - заголовок, 2 - с учетом индекса
                    for cell in depth_ws[excel_row]:
                        cell.font = Font(bold=True)

    output.seek(0)

    await message.answer_document(
        types.BufferedInputFile(
            file=output.read(),
            filename=f"orderbook_{exchange if exchange else 'all'}.xlsx"
        ),
        caption=f"Excel file with actual orders{' for ' + exchange if exchange else ''}"
    )


@dp.message(F.text.lower().contains('add_token_'))
async def handler(message: Message):
    current_chat_id = str(message.chat.id)
    user_id = message.from_user.id

    if str(user_id) not in admin_ids:
        await message.reply(f"you haven't admin rights to add token")
        return

    if '.' not in message.text:
        token_name = message.text.split('_')[2].upper()
    else:
        token_name = '_'.join(message.text.split('_')[2:])
        token_name = token_name.split('.')[0].upper() + '.' + token_name.split('.')[1].lower()

    if '<' in token_name and '>' in token_name:
        token_name = token_name[1:-1]

    if current_chat_id in chat_ids:
        if token_name in chat_ids[current_chat_id]:
            await message.reply(f"{token_name} have already been added")
            return
        chat_ids[current_chat_id].append(token_name)
    else:
        chat_ids[current_chat_id] = [token_name]

    with open('clients.json', 'w') as clients_f:
        clients_f.write(json.dumps(chat_ids))

    await dbReader.init_pool()
    await dbSideCashoutReader.init_pool(side=True)

    await dbReader.update_symbols_list()
    await dbSideCashoutReader.update_symbols_list(side=True)

    await message.reply(f"{token_name} was successfully added")


@dp.message(F.text.lower().contains('delete_token_'))
async def handler(message: Message):
    current_chat_id = str(message.chat.id)
    user_id = message.from_user.id

    if str(user_id) not in admin_ids:
        await message.reply(f"you haven't admin rights to add token")
        return

    if '.' not in message.text:
        token_name = message.text.split('_')[2].upper()
    else:
        token_name = '_'.join(message.text.split('_')[2:])
        token_name = token_name.split('.')[0].upper() + '.' + token_name.split('.')[1].lower()

    if '<' in token_name and '>' in token_name:
        token_name = token_name[1:-1]

    if current_chat_id in chat_ids:
        if token_name not in chat_ids[current_chat_id]:
            await message.reply(f"{token_name} not in tokens list")
            return
        chat_ids[current_chat_id].remove(token_name)
    else:
        await message.reply(f"Your chat is not in chat ids")
        return

    with open('clients.json', 'w') as clients_f:
        clients_f.write(json.dumps(chat_ids))

    await dbReader.init_pool()
    await dbSideCashoutReader.init_pool(side=True)

    await dbReader.update_symbols_list()
    await dbSideCashoutReader.update_symbols_list(side=True)

    await message.reply(f"{token_name} was successfully deleted")



@dp.message(Command("conditions"))
async def market_conditions(message: types.Message):
    current_chat_id = str(message.chat.id)
    await dbReader.init_pool()

    result = ''

    for group in chat_ids.get(current_chat_id, []):
        for symbol_id in dbReader.symbols_id_in_group.get(group, []):
            try:
                symbol_data = await dbReader.execute_select(
                    f"""SELECT concat(exchange_name, \'.\', exchange_type) AS exchange_name, symbol, price_prec, qty_prec, hostname
                        FROM symbols
                        JOIN exchanges USING (exchange_id)
                        JOIN id_group_relation USING (symbol_id)
                        JOIN group_host_relation USING (group_id)
                        JOIN hosts USING (host_id)
                        WHERE symbol_id = {symbol_id}
                    """
                )

                if not symbol_data:
                    continue

                symbol_data = symbol_data[0]

                # TODO market_symbol if market_symbol != {} else symbol_data.symbol ????
                res = await manual_get_ubook2_async(
                    symbol_data[4], # hostname
                    symbol_data[0], # exchange_name
                    symbol_data[1], # symbol
                    (symbol_data[2], symbol_data[3]) # decimals
                )
                if not res:
                    continue
                ask_p, bid_p, uask2, ubid2 = res[0], res[1], res[2], res[3]
                spread = round(100*(ask_p - bid_p) / bid_p, 2) if bid_p != 0 else 'None'

                resp = (
                    f"{symbol_data[1]} {symbol_data[0].replace('.', ' ')}\n"
                    f"🔴 Ask: {ask_p}\n"
                    f"🟢 Bid: {bid_p}\n"
                    f"➖ Spread: {spread}%\n"
                    f"🔼 Depth +2%: {uask2}\n"
                    f"🔽 Depth -2%: {ubid2}\n\n"
                )

                result += resp

            except Exception as e:
                print(f'[Exception] market_conditions {group} {symbol_id}: {e}')

    if result:
        await message.answer(result)


###############################################################
######################## ALARM BOT ############################
###############################################################

@dp.message(Command("start"))
async def start(message: types.Message):
    if str(message.from_user.id) not in admin_ids:
        await message.reply(f"you haven't admin rights to start bot")
        return

    title = message.chat.title
    if title.find('<') == -1 or title.find('>') == -1 or title.find('<') >= title.find('>'):
        await message.answer("Failed: wrong chat name")
        return

    token = title[(title.find('<')+1):title.find('>')].upper()
    chats_mapping[token] = str(message.chat.id)

    with open('/home/ubuntu/tgbot/chat-id-token.json', 'w') as token_f:
        token_f.write(json.dumps(chats_mapping))

    await message.answer(f'Your chat title: \"{title}\"\nYour token: {token}')


@dp.message(Command("stop"))
async def stop(message: types.Message):
    if str(message.from_user.id) not in admin_ids:
        await message.reply(f"you haven't admin rights to stop bot")
        return

    # TODO: сделать чтобы несколько чатов могли иметь одинаковый токен
    tokens_to_delete = [t for t, cid in chats_mapping.items() if cid == str(message.chat.id)]

    if len(tokens_to_delete) == 0:
        await message.answer(f'Your chat already was deleted from alarms list')
        return

    for t in tokens_to_delete:
        chats_mapping.pop(t, None)

    with open('/home/ubuntu/tgbot/chat-id-token.json', 'w') as token_f:
        token_f.write(json.dumps(chats_mapping))

    await message.answer(f'Your chat with token {tokens_to_delete[0]} was successfully deleted from alarms list')


alarms = {}

async def manage_alarm(bot, chat_id, value, alarm_type):
    try:  # на всякий случай
        symbol = value['symbol']
        if symbol not in alarms:
            alarms[symbol] = {
                'aliveness': {
                    'status': True,
                    'ts': 0  # last alarm time
                },
                'spread': {
                    'status': True,
                    'prev_values': [],
                    'ts': 0  # last alarm time
                },
                'flat': {
                    'status': True,       # armed: non-flat candle seen since last alert
                    'last_bucket': None,  # epoch of the candle we last alerted on
                    'ts': 0               # last alarm time (2h cooldown)
                }
            }

        info = alarms[symbol]

        if alarm_type == 'spread':
            if int(time.time()) - info['spread']['ts'] > 60 * 15:
                if value['spread'] > 0.03 and info['spread']['status']:
                    try:
                        await bot.send_message(
                            text=f"{symbol}:\n--- spread > 3% ---\n~~~ ask/bid/spread   {value['ask_price']} / {value['bid_price']} / {round(100*value['spread'], 2)}%",
                            chat_id=chat_id
                        )
                    except Exception as e:
                        print('exception', e)

                info['spread']['ts'] = int(time.time())
                info['spread']['status'] = (value['spread'] <= 0.03)

        elif alarm_type == 'aliveness':
            if int(time.time()) - info['aliveness']['ts'] > 60 * 15:
                if not value['status'] and info['aliveness']['status']:
                    try:
                        await bot.send_message(
                            text=f"{symbol}:\n--- DEAD ---",
                            chat_id=chat_id
                        )
                    except Exception as e:
                        print('exception', e)

                info['aliveness']['ts'] = int(time.time())
                info['aliveness']['status'] = value['status']

        elif alarm_type == 'flat':
            # Two layers of anti-spam, gated by a 2h cooldown:
            #   - cooldown: at most one flat alert per symbol every 2 hours;
            #   - arming: within that, alert only on a *fresh* flat episode —
            #     fire while armed, then disarm; re-arm on a non-flat reading
            #     (the market moved). 'last_bucket' guards against re-alerting
            #     the very same candle.
            # Candles with < 2 trades never reach here: get_flat_candles_by_group
            # excludes them, so a single-trade window is treated as "no data".
            fi = info['flat']
            bucket = value['last_bucket']
            if int(time.time()) - fi['ts'] > 60 * 60 * 2:  # 2h cooldown
                if value['flat'] and fi['status'] and fi['last_bucket'] != bucket:
                    try:
                        await bot.send_message(
                            text=f"{symbol}:\n--- FLAT ---\n~~~ last two 15m candles closed flat (open == close, ≥2 trades each)",
                            chat_id=chat_id
                        )
                    except Exception as e:
                        print('exception', e)
                    fi['last_bucket'] = bucket
                    fi['status'] = False
                    fi['ts'] = int(time.time())
                elif not value['flat']:
                    # market moved — re-arm for the next flat episode
                    fi['status'] = True
                    fi['last_bucket'] = bucket

    except Exception as e:
        print(f'EXCEPTION: manage_alarm {value} {alarm_type} {e}')


async def book_checker(bot):
    await dbReader.init_pool()

    for group, chat_id in chats_mapping.items():
        if debug and chat_id != '-1002388231434': # debug mock
            continue
        try:
            books = await dbReader.get_book_by_group(group)
        except Exception as e:
            print(f'EXCEPTION: book_checker {books} - {e}')
            continue
        for book in books:
            ts = int(datetime.fromisoformat(book['time']).timestamp())
            if int(time.time()) - ts <= 60:
                await manage_alarm(bot, chat_id, book, 'spread')
            else:
                print('NO ACTUAL BOOK', int(time.time()), group, book)


async def flat_checker(bot):
    await dbReader.init_pool()

    for group, chat_id in chats_mapping.items():
        if debug and chat_id != '-1002388231434':  # debug mock
            continue
        try:
            candles = await dbReader.get_flat_candles_by_group(group)
        except Exception as e:
            print(f'EXCEPTION: flat_checker {group} - {e}')
            continue
        for candle in candles:
            await manage_alarm(bot, chat_id, candle, 'flat')


async def status_checker(bot):
    await dbReader.init_pool()
    statuses = await dbReader.get_last_statuses()

    for group, chat_id in chats_mapping.items():
        if debug and chat_id != '-1002388231434': # debug mock
            continue
        if group not in dbReader.symbols_id_in_group:
            continue
        for symbol in dbReader.symbols_id_in_group[group]:
            status_v, status_m = None, None

            if symbol not in dbReader.symbol_id_to_ticket:
                continue

            if symbol in statuses:
                for status in statuses[symbol]:
                    if status['strategy'] == 'M':
                        status_m = status['status']
                    if status['strategy'] == 'V':
                        status_v = status['status']

            symbol = dbReader.symbol_id_to_ticket[symbol]
            symbol_v = f'{symbol} Volume Bot'
            symbol_m = f'{symbol} Market Making Bot'

            await manage_alarm(bot, chat_id, {'symbol': symbol_v, 'status': status_v}, 'aliveness')
            await manage_alarm(bot, chat_id, {'symbol': symbol_m, 'status': status_m}, 'aliveness')

###############################################################
###############################################################


###############################################################
##################### LOW BALANCE ALERTER #####################
###############################################################

LOW_BALANCE_THRESHOLD_USDT = 150

STRATEGY_LABELS = {
    'V': 'Volume',
    'M': 'Market Making',
    'T': 'Treasury Building',
    'W': 'BBO',
}


async def _current_price(reader, symbol_id):
    """Best-effort current price: book midpoint, fallback to last public trade."""
    rows = await reader.execute_select(f"""
        SELECT ask_price, bid_price
        FROM book_{symbol_id}
        WHERE ask_price > 0 AND bid_price > 0
        ORDER BY time DESC
        LIMIT 1
    """)
    if rows and rows[0] and rows[0][0]:
        return (float(rows[0][0]) + float(rows[0][1])) / 2
    rows = await reader.execute_select(f"""
        SELECT price
        FROM public_trade_{symbol_id}
        WHERE price > 0
        ORDER BY time DESC
        LIMIT 1
    """)
    if rows and rows[0]:
        return float(rows[0][0])
    return 0


def _format_low_balance_alert(strategy, market, token, base, quote):
    strat_label = STRATEGY_LABELS.get(strategy, strategy)
    return (
        f"⚠️ <b>ATTENTION on balances</b>\n"
        f"<b>Strategy:</b> {strat_label}\n"
        f"<b>Market:</b> {market}\n"
        f"<b>Current balances:</b>\n"
        f"{token} — <code>{round(float(base), 4)}</code>\n"
        f"USDT — <code>{round(float(quote), 2)}</code>"
    )


async def low_balance_alerter(bot):
    """Daily check: send an alert to each chat for every (symbol, strategy)
    where quote + base*price < LOW_BALANCE_THRESHOLD_USDT."""
    try:
        await dbReader.init_pool()
        await dbSideCashoutReader.init_pool(side=True)
        await dbReader.update_symbols_list()
        await dbSideCashoutReader.update_symbols_list(side=True)
    except Exception as e:
        print(f'[low_balance_alerter] init failed: {e}')
        return

    # Walks the same chats as DEAD / spread alarms — chats_mapping
    # ({token: chat_id}, filled by /start). Side cashout tokens are skipped.
    for token, chat_id in chats_mapping.items():
        if '.' in token:
            continue
        symbols = dbReader.symbols_id_in_group.get(token, [])
        alert_blocks = []

        for symbol in symbols:
            market = dbReader.symbol_id_to_ticket.get(symbol, str(symbol))
            base_token = market.split('/')[0]

            try:
                price = await _current_price(dbReader, symbol)
            except Exception as e:
                print(f'[low_balance_alerter] price fetch failed {symbol}: {e}')
                price = 0

            # Only V (Volume) and M (Market Making) balances are tracked here;
            # treasury / cashout strategies are intentionally skipped.
            rows = await dbReader.execute_select(f"""
                SELECT DISTINCT ON (strategy) strategy, base, quote
                FROM balance_{symbol}
                WHERE strategy IN ('V', 'M')
                  AND time > NOW() - interval '7 days'
                ORDER BY strategy, time DESC
            """)
            if not rows:
                continue
            for strat, base_v, quote_v in rows:
                base = float(base_v)
                quote = float(quote_v)
                total = quote + base * price
                if total < LOW_BALANCE_THRESHOLD_USDT:
                    alert_blocks.append(
                        _format_low_balance_alert(strat, market, base_token, base, quote)
                    )

        if not alert_blocks:
            continue

        text = '\n\n'.join(alert_blocks)
        try:
            for chunk in split_html_safe(text):
                await bot.send_message(chat_id, chunk, parse_mode=ParseMode.HTML)
        except Exception as e:
            print(f'[low_balance_alerter] send failed chat={chat_id}: {e}')


@dp.message(F.text)
async def bad_message(message: Message): # нельзя поднимать эту функцию выше!
    pass


async def main():
    scheduler = AsyncIOScheduler()
    if not debug:
        scheduler.add_job(book_checker, "interval", seconds=75, args=[bot])
        scheduler.add_job(status_checker, "interval", seconds=60, args=[bot])
        scheduler.add_job(flat_checker, "interval", seconds=300, args=[bot])
        scheduler.add_job(
            low_balance_alerter, "cron",
            hour=15, minute=30, timezone="UTC", args=[bot],
        )
    scheduler.start()

    await dp.start_polling(bot)
    await dbReader.close_connection()

if __name__ == "__main__":
    asyncio.run(main())