модуль 3

import os import sys import json import random import sqlite3 import argparse import asyncio from pathlib import Path from dataclasses import dataclass, asdict, field from typing import Dict, List, Tuple, Optional import aiohttp from openai import OpenAI from web3 import Web3 from bs4 import BeautifulSoup # --- СТАТИЧЕСКАЯ КОНФИГУРАЦИЯ И СТАНДАРТЫ СИСТЕМЫ --- TARGET_MARGIN = 0.50 # Целевая маржинальность ИИ (50%) USDC_CONTRACT_ADDRESS = "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174" # Сеть Polygon/Base # Инициализация внешних провайдеров (при их отсутствии в ENV используются безопасные заглушки) OPENAI_KEY = os.getenv("OPENAI_API_KEY", "mock_key") WEB3_PROVIDER = os.getenv("RPC_PROVIDER_URL", "https://polygon-rpc.com") openai_client = OpenAI(api_key=OPENAI_KEY) w3 = Web3(Web3.HTTPProvider(WEB3_PROVIDER)) MIN_ERC20_ABI = [ {"constant": True, "inputs": [{"name": "_owner", "type": "address"}], "name": "balanceOf", "outputs": [{"name": "balance", "type": "uint256"}], "type": "function"}, {"constant": True, "inputs": [], "name": "decimals", "outputs": [{"name": "remaining", "type": "uint8"}], "type": "function"} ] @dataclass class ChannelState: name: str expected_aov: float alpha_sales: float = 1.0 # Априорное Бета-распределение для Сэмплирования Томпсона beta_failures: float = 1.0 total_clicks: int = 0 total_sales: int = 0 total_revenue: float = 0.0 total_cost: float = 0.0 # ===================================================================== # 1. МАТЕМАТИЧЕСКОЕ ЯДРО ПРИНЯТИЯ РЕШЕНИЙ (THOMPSON SAMPLING OPTIMIZER) # ===================================================================== class AdvancedOptimizer: def __init__(self): self.channels: Dict[str, ChannelState] = {} self.history_file = "bid_history.json" self.performance_memory: List[dict] = self._load_memory() def _load_memory(self) -> List[dict]: if Path(self.history_file).exists(): try: return json.loads(Path(self.history_file).read_text(encoding="utf-8")) except Exception: return [] return [] def save_state(self, path: Path): serializable = {k: asdict(v) for k, v in self.channels.items()} path.write_text(json.dumps(serializable, indent=2, ensure_ascii=False), encoding="utf-8") @classmethod def load_state(cls, path: Path): opt = cls() if path.exists(): data = json.loads(path.read_text(encoding="utf-8")) for k, v in data.items(): opt.channels[k] = ChannelState(**v) return opt def get(self, name: str, aov: float): if name not in self.channels: self.channels[name] = ChannelState(name=name, expected_aov=aov) def observe(self, name: str, clicks: int, sales: int, revenue: float, cost: float, aov: Optional[float] = None): self.get(name, aov or 60.0) ch = self.channels[name] ch.total_clicks += clicks ch.total_sales += sales ch.total_revenue += revenue ch.total_cost += cost # Обновление байесовских параметров обучения (Успех / Неуспех) ch.alpha_sales += sales ch.beta_failures += max(0, clicks - sales) if aov: ch.expected_aov = (ch.expected_aov * 0.7) + (aov * 0.3) # Скользящее среднее чека def decide(self, total_budget: float) -> dict: """Распределяет бюджет на основе сэмплирования Томпсона и максимизации маржи""" allocation = [] sampled_rates = {} # Шаг 1: Математическое сэмплирование вероятности конверсии из Бета-распределения for name, ch in self.channels.items(): sampled_p = random.betavariate(ch.alpha_sales, ch.beta_failures) sampled_rates[name] = sampled_p # Фильтруем каналы с положительным математическим ожиданием прибыли valid_channels = [] for name, p_est in sampled_rates.items(): ch = self.channels[name] # Примерная рыночная ставка (оценка на основе прошлых затрат или дефолт) est_cpc = (ch.total_cost / ch.total_clicks) if ch.total_clicks > 0 else 0.50 expected_revenue_per_click = p_est * ch.expected_aov expected_margin = (expected_revenue_per_click - est_cpc) / expected_revenue_per_click if expected_revenue_per_click > 0 else -1 if expected_margin >= 0.10: # Берем только потенциально прибыльные valid_channels.append((name, expected_margin, est_cpc, p_est)) if not valid_channels: # Если всё плохо, распределяем поровну между всеми известными в качестве Exploration share = total_budget / len(self.channels) if self.channels else total_budget for name in self.channels: allocation.append({ "channel": name, "budget": share, "bidperclick": 0.50, "status": "EXPLORE", "pconv": 0.01, "cacfact": "N/A", "marginfact": "N/A" }) return {"allocation": allocation} # Шаг 2: Пропорциональное распределение бюджета по весам маржинальности total_margin_weight = sum(m for _, m, _, _ in valid_channels) for name, margin, cpc, p_est in valid_channels: weight = margin / total_margin_weight allocated_money = total_budget * weight # Реальные показатели эффективности для вывода консоли ch = self.channels[name] real_cac = (ch.total_cost / ch.total_sales) if ch.total_sales > 0 else 0.0 real_margin = (ch.total_revenue - ch.total_cost) / ch.total_revenue if ch.total_revenue > 0 else 0.0 allocation.append({ "channel": name, "budget": round(allocated_money, 2), "bidperclick": round(cpc, 2), "status": "ACTIVE_MAX", "pconv": f"{p_est:.1%}", "cacfact": f"{real_cac:.1f}", "marginfact": f"{real_margin:.1%}" }) return {"allocation": allocation} # ===================================================================== # 2. МОДУЛЬ ГЛОБАЛЬНЫХ ИИ-ОПЕРАЦИЙ (WEB3, LLM JSON METADATA, OUTREACH) # ===================================================================== class AutonomousAIEngine: def __init__(self, wallet_address: str, product_desc: str, target_aud: str): self.wallet_address = wallet_address self.product_desc = product_desc self.target_aud = target_aud def check_blockchain_balance(self) -> float: try: contract = w3.eth.contract(address=w3.to_checksum_address(USDC_CONTRACT_ADDRESS), abi=MIN_ERC20_ABI) raw_balance = contract.functions.balanceOf(w3.to_checksum_address(self.wallet_address)).call() decimals = contract.functions.decimals().call() return raw_balance / (10 ** decimals) except Exception: print("[⚠️ Web3] RPC unavailable. Using Simulated Sandbox Balance.") return 500.0 # Песочный баланс по умолчанию def generate_structured_campaign(self) -> dict: """Интеграция OpenAI Structured Outputs с жесткой валидацией JSON-схемы""" if OPENAI_KEY == "mock_key": return { "ad_text": "Mock Promo: Scale your crypto business automatically!", "banner_prompt": "Cyberpunk server room", "estimated_cpc_target": 0.45, "suggested_keywords": ["crypto", "saas"] } json_schema = { "name": "ad_campaign_config", "strict": True, "schema": { "type": "object", "properties": { "ad_text": {"type": "string", "description": " Telegram promotional text with emojis."}, "banner_prompt": {"type": "string", "description": "DALL-E 3 minimalistic aesthetic prompt."}, "estimated_cpc_target": {"type": "number", "description": "Max optimal CPC threshold."}, "suggested_keywords": {"type": "array", "items": {"type": "string"}, "description": "Top 2 niche search queries."} }, "required": ["ad_text", "banner_prompt", "estimated_cpc_target", "suggested_keywords"], "additionalProperties": False } } response = openai_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "You are a decentralized self-optimizing media buying node."}, {"role": "user", "content": f"Product: {self.product_desc}\nAudience: {self.target_aud}"} ], response_format={"type": "json_schema", "json_schema": json_schema} ) return json.loads(response.choices.message.content) # ===================================================================== # 3. ДИНАМИЧЕСКИЙ СИМУЛЯТОР РЫНКА (STRESS-TEST BENCHMARK) # ===================================================================== def run_market_simulation(rounds: int): random.seed(42) truth = { "chana": [0.030, 55.0, 0.35], "chanb": [0.012, 90.0, 0.90], "chanc": [0.006, 140.0, 1.60] } opt = AdvancedOptimizer() budget_per_round = 100.0 for name, (_, aov, _) in truth.items(): opt.get(name, aov) block = {"spend": 0.0, "revenue": 0.0} log = [] for r in range(1, rounds + 1): # Триггеры изменения рынка (Смена тренда) if r == 301: truth["chana"][0] = 0.008 # Прежний лидер резко просел if r == 351: truth["chand"] = [0.025, 70.0, 0.50] # Появился новый сильный игрок opt.get("chand", 70.0) decision = opt.decide(budget_per_round)