#!/usr/bin/env python3 # ═══════════════════════════════════════════════════════════════ # 🎮 Craft World - Economy v1.0.0 # Bot de Alertas de Preços + Custo de Produção # # Desenvolvido por Plum com Qwen # Telegram: @bondsbtc # ═══════════════════════════════════════════════════════════════ import json import time import logging import requests from datetime import datetime, timedelta from config import * from utils import * from telegram_bot import * # Configurar logging logger = setup_logging(LOG_FILE, LOG_LEVEL) def get_batch_prices(pool_addresses, network=NETWORK): """Busca preços de múltiplas pools em UMA chamada""" valid_addresses = [addr for addr in pool_addresses if addr and addr.startswith("0x") and "PENDENTE" not in addr] if not valid_addresses: logger.warning("Nenhum endereço de pool válido") return {} all_prices = {} for i in range(0, len(valid_addresses), MAX_BATCH_SIZE): batch = valid_addresses[i:i + MAX_BATCH_SIZE] addresses = ",".join(batch) url = f"{GECKO_BASE_URL}/networks/{network}/pools/multi/{addresses}" try: response = requests.get(url, headers=API_HEADERS, timeout=30) response.raise_for_status() data = response.json() for item in data.get('data', []): pool_id = item['id'] all_prices[pool_id] = { 'price_usd': float(item['attributes']['price_usd'] or 0), 'volume_usd_24h': float(item['attributes']['volume_usd_h24'] or 0), 'price_change_24h': float(item['attributes']['price_change_percentage_h24'] or 0), 'updated_at': item['attributes']['updated_at'], 'fetched_at': datetime.utcnow().isoformat() } if i + MAX_BATCH_SIZE < len(valid_addresses): time.sleep(RATE_LIMIT_DELAY) except requests.exceptions.RequestException as e: logger.error(f"Erro na API GeckoTerminal: {e}") except (KeyError, ValueError) as e: logger.error(f"Erro ao processar resposta: {e}") return all_prices def load_price_cache(): """Carrega cache de preços""" cache = load_json_file(PRICE_CACHE_FILE) if cache and 'timestamp' in cache: age = time.time() - cache['timestamp'] if age < PRICE_CACHE_TTL_SECONDS: logger.info(f"Usando cache (idade: {age:.0f}s)") return cache.get('prices', {}) return {} def save_price_cache(prices): """Salva cache de preços""" cache = { 'timestamp': time.time(), 'prices': prices, 'count': len(prices) } save_json_file(PRICE_CACHE_FILE, cache) logger.info(f"Cache salvo ({len(prices)} pools)") def analyze_opportunities(market_prices, production_costs, alerts_config, language=None): """Analisa oportunidades de compra/venda""" if language is None: language = DEFAULT_LANGUAGE opportunities = [] global_config = alerts_config.get('global', {}) resources_config = alerts_config.get('resources', {}) buy_threshold = global_config.get('buy_threshold_percent', 15) sell_threshold = global_config.get('sell_threshold_percent', 20) pools = load_json_file(POOLS_FILE) address_to_symbol = {v: k for k, v in pools.items()} for pool_address, price_data in market_prices.items(): symbol = address_to_symbol.get(pool_address) if not symbol or symbol not in resources_config: continue cfg = resources_config[symbol] if not cfg.get('enabled', False): continue market_price = price_data['price_usd'] prod_cost = calculate_production_cost(symbol, production_costs, market_prices) if market_price <= 0 or prod_cost <= 0: continue deviation = ((market_price - prod_cost) / prod_cost) * 100 alert_type = None if deviation < -buy_threshold: alert_type = "OPORTUNIDADE_DE_COMPRA" if language == 'pt' else "BUY_OPPORTUNITY" elif deviation > sell_threshold: alert_type = "OPORTUNIDADE_DE_VENDA" if language == 'pt' else "SELL_OPPORTUNITY" if alert_type: opportunities.append({ 'symbol': symbol, 'market_price': market_price, 'production_cost': prod_cost, 'deviation_percent': deviation, 'type': alert_type, 'priority': cfg.get('priority', 'medium'), 'timestamp': datetime.utcnow().isoformat(), 'category': cfg.get('category', 'unknown') }) return opportunities def check_alert_cooldown(symbol, alerts_config, cooldown_minutes=30): """Verifica se alerta está em cooldown""" last_alerts = alerts_config.get('last_alerts', {}) if symbol in last_alerts: last_time = datetime.fromisoformat(last_alerts[symbol]) if datetime.utcnow() - last_time < timedelta(minutes=cooldown_minutes): return True return False def main(): """Função principal""" language = DEFAULT_LANGUAGE logger.info("=" * 60) logger.info(f"{PROJECT_NAME} v{VERSION} - Iniciando") logger.info(f"Desenvolvido por {DEVELOPER}") logger.info("=" * 60) pools = load_json_file(POOLS_FILE) production_costs = load_json_file(PRODUCTION_COSTS_FILE) alerts_config = load_json_file(ALERTS_CONFIG_FILE) if not pools: logger.error("Nenhuma pool configurada") return active_pools = [addr for addr in pools.values() if addr and addr.startswith("0x")] logger.info(f"Pools ativas: {len(active_pools)}") cached_prices = load_price_cache() if cached_prices: prices = cached_prices else: logger.info("Buscando preços na API...") prices = get_batch_prices(active_pools) if prices: save_price_cache(prices) if not prices: logger.warning("Nenhum preço obtido") return logger.info(f"Preços atualizados: {len(prices)}") opportunities = analyze_opportunities(prices, production_costs, alerts_config, language) if opportunities: logger.warning(f"🚨 {len(opportunities)} oportunidade(s) encontrada(s)!") for opp in opportunities: if check_alert_cooldown(opp['symbol'], alerts_config): logger.info(f"Alerta em cooldown: {opp['symbol']}") continue send_price_alert(opp, language) alerts_config['last_alerts'][opp['symbol']] = datetime.utcnow().isoformat() logger.warning( f"{opp['type']}: {opp['symbol']} - " f"Mercado: ${opp['market_price']:.8f} - " f"Custo: ${opp['production_cost']:.8f} - " f"Desvio: {opp['deviation_percent']:.1f}%" ) save_json_file(ALERTS_CONFIG_FILE, alerts_config) else: logger.info("✅ Sem oportunidades no momento") logger.info(f"Concluído - {datetime.utcnow().isoformat()}") logger.info("=" * 60) if __name__ == "__main__": if not load_json_file(PRICE_CACHE_FILE): logger.info("Primeira execução - enviando teste") send_welcome_message(DEFAULT_LANGUAGE) main()