import sqlite3 from datetime import datetime class AetherisMemory: def __init__(self, db_path="aetheris_memory.db"): import os from pathlib import Path if db_path == "aetheris_memory.db": local_runtime = Path("/Users/blacksun/.aetheris/src") if local_runtime.exists(): db_path = str(local_runtime / "aetheris_memory.db") else: db_path = str(Path(__file__).parent.resolve() / "aetheris_memory.db") elif not os.path.isabs(db_path): db_path = str(Path(__file__).parent.resolve() / db_path) self.db_path = db_path self._create_tables() def _get_connection(self): return sqlite3.connect(self.db_path, timeout=30.0) def _create_tables(self): with self._get_connection() as conn: conn.execute("PRAGMA journal_mode=WAL;") # Bảng lưu trải nghiệm (dữ liệu thô để Train) conn.execute(""" CREATE TABLE IF NOT EXISTS experiences ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME, coin TEXT, obi REAL, funding REAL, prediction REAL, actual_outcome REAL DEFAULT 0, processed INTEGER DEFAULT 0, signals_json TEXT, label INTEGER DEFAULT 0, source TEXT DEFAULT 'live' ) """) # Đảm bảo các cột bổ sung tồn tại (migration cho db cũ) columns = [c[1] for c in conn.execute("PRAGMA table_info(experiences)").fetchall()] if "coin" not in columns: conn.execute("ALTER TABLE experiences ADD COLUMN coin TEXT;") if "signals_json" not in columns: conn.execute("ALTER TABLE experiences ADD COLUMN signals_json TEXT;") if "label" not in columns: conn.execute("ALTER TABLE experiences ADD COLUMN label INTEGER DEFAULT 0;") if "source" not in columns: conn.execute("ALTER TABLE experiences ADD COLUMN source TEXT DEFAULT 'live';") # Bảng lưu lịch sử giao dịch (để Evaluate) conn.execute(""" CREATE TABLE IF NOT EXISTS trades ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME, coin TEXT, side TEXT, entry_px REAL, exit_px REAL, pnl REAL ) """) # Bảng lưu Live Signals để show lên Dashboard conn.execute(""" CREATE TABLE IF NOT EXISTS live_signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at DATETIME, coin TEXT, side TEXT, entry_price REAL, sl_price REAL, tp_price REAL, order_type TEXT, confidence REAL, status TEXT, closed_at DATETIME, final_pnl REAL ) """) # Bảng lưu thông tin tiếp thị quảng cáo chợ Bazaar conn.execute(""" CREATE TABLE IF NOT EXISTS bazaar_promotions ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at DATETIME, repo_name TEXT UNIQUE, repo_url TEXT, description TEXT, pitch_draft TEXT, developer_email TEXT, email_pitch TEXT, status TEXT DEFAULT 'PENDING', shipped_at DATETIME ) """) # Migration check for existing bazaar_promotions promo_cols = [c[1] for c in conn.execute("PRAGMA table_info(bazaar_promotions)").fetchall()] if "developer_email" not in promo_cols: conn.execute("ALTER TABLE bazaar_promotions ADD COLUMN developer_email TEXT;") if "email_pitch" not in promo_cols: conn.execute("ALTER TABLE bazaar_promotions ADD COLUMN email_pitch TEXT;") # Bảng lưu các dịch vụ Agent đăng ký trên chợ Bazaar conn.execute(""" CREATE TABLE IF NOT EXISTS bazaar_services ( id INTEGER PRIMARY KEY AUTOINCREMENT, registered_at DATETIME, service_name TEXT UNIQUE, endpoint TEXT, price_gwei REAL, description TEXT, wallet TEXT ) """) # Bảng lưu giao dịch thanh toán đã xác thực (tránh double spend) conn.execute(""" CREATE TABLE IF NOT EXISTS bazaar_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, redeemed_at DATETIME, tx_hash TEXT UNIQUE, service_name TEXT ) """) # ── EXPERIENCES ─────────────────────────────────────────── def add_experience(self, obi, funding, prediction, coin=None, signals_json=None, label=None, source="live"): """Ghi lại tín hiệu vừa phát sinh để học sau. Trả về ID để feedback loop.""" with self._get_connection() as conn: cursor = conn.execute( "INSERT INTO experiences (timestamp, coin, obi, funding, prediction, signals_json, label, source) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (datetime.now().isoformat(), coin, obi, funding, prediction, signals_json, label, source) ) return cursor.lastrowid def update_outcome(self, price_change): """Ghi nhận biến động giá thực tế cho experience gần nhất chưa xử lý.""" with self._get_connection() as conn: conn.execute( "UPDATE experiences SET actual_outcome = ? " "WHERE processed = 0 AND id = (SELECT MAX(id) FROM experiences WHERE processed = 0)", (price_change,) ) def get_unprocessed(self, limit=200): """Lấy các experiences chưa được học.""" with self._get_connection() as conn: return conn.execute( "SELECT id, obi, funding, prediction, actual_outcome " "FROM experiences WHERE processed = 0 LIMIT ?", (limit,) ).fetchall() def mark_as_processed(self, ids): """Đánh dấu đã học xong.""" with self._get_connection() as conn: conn.executemany( "UPDATE experiences SET processed = 1 WHERE id = ?", [(i,) for i in ids] ) # ── TRADES ──────────────────────────────────────────────── def add_trade(self, coin, side, entry_px, exit_px, pnl): """Ghi nhận kết quả một lệnh giao dịch.""" with self._get_connection() as conn: conn.execute( "INSERT INTO trades (timestamp, coin, side, entry_px, exit_px, pnl) " "VALUES (?, ?, ?, ?, ?, ?)", (datetime.now().isoformat(), coin, side, entry_px, exit_px, pnl) ) def get_recent_trades(self, limit=50): """Lấy N lệnh gần nhất.""" with self._get_connection() as conn: return conn.execute( "SELECT id, timestamp, coin, side, entry_px, exit_px, pnl " "FROM trades ORDER BY id DESC LIMIT ?", (limit,) ).fetchall() # ── LIVE SIGNALS ────────────────────────────────────────── def add_live_signal(self, coin, side, entry_price, sl_price, tp_price, order_type, confidence): with self._get_connection() as conn: cursor = conn.execute( """INSERT INTO live_signals (created_at, coin, side, entry_price, sl_price, tp_price, order_type, confidence, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (datetime.now().isoformat(), coin, side, float(entry_price), float(sl_price), float(tp_price), order_type, float(confidence), "PENDING") ) return cursor.lastrowid def update_live_signal(self, signal_id, status, final_pnl): with self._get_connection() as conn: conn.execute( """UPDATE live_signals SET status = ?, closed_at = ?, final_pnl = ? WHERE id = ?""", (status, datetime.now().isoformat(), final_pnl, signal_id) ) def get_recent_live_signals(self, limit=10): with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute("SELECT * FROM live_signals ORDER BY id DESC LIMIT ?", (limit,)).fetchall() return [dict(row) for row in rows] # ── BAZAAR PROMOTIONS ───────────────────────────────────── def add_bazaar_promotion(self, repo_name, repo_url, description, pitch_draft): with self._get_connection() as conn: try: conn.execute( """INSERT INTO bazaar_promotions (created_at, repo_name, repo_url, description, pitch_draft, status) VALUES (?, ?, ?, ?, ?, ?)""", (datetime.now().isoformat(), repo_name, repo_url, description, pitch_draft, "PENDING") ) return True except sqlite3.IntegrityError: # Repo already exists return False def get_pending_promotions(self, limit=10): with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT * FROM bazaar_promotions WHERE status = 'PENDING' ORDER BY id DESC LIMIT ?", (limit,) ).fetchall() return [dict(row) for row in rows] def update_promotion_status(self, promo_id, status): with self._get_connection() as conn: conn.execute( "UPDATE bazaar_promotions SET status = ?, shipped_at = ? WHERE id = ?", (status, datetime.now().isoformat() if status == "APPROVED" or status == "SHIPPED" else None, promo_id) ) # ── BAZAAR SERVICES & TRANSACTIONS ─────────────────────── def add_bazaar_service(self, service_name, endpoint, price_gwei, description, wallet): with self._get_connection() as conn: try: conn.execute( """INSERT INTO bazaar_services (registered_at, service_name, endpoint, price_gwei, description, wallet) VALUES (?, ?, ?, ?, ?, ?)""", (datetime.now().isoformat(), service_name.strip(), endpoint.strip(), float(price_gwei), description, wallet.strip()) ) return True except sqlite3.IntegrityError: return False def get_bazaar_services(self): with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute("SELECT * FROM bazaar_services ORDER BY id DESC").fetchall() return [dict(row) for row in rows] def verify_and_redeem_tx(self, tx_hash, service_name): with self._get_connection() as conn: try: conn.execute( """INSERT INTO bazaar_transactions (redeemed_at, tx_hash, service_name) VALUES (?, ?, ?)""", (datetime.now().isoformat(), tx_hash.strip(), service_name.strip()) ) return True except sqlite3.IntegrityError: # Double spend detected (tx_hash already redeemed) return False def get_bazaar_transactions(self, limit=10): with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute("SELECT * FROM bazaar_transactions ORDER BY id DESC LIMIT ?", (limit,)).fetchall() return [dict(row) for row in rows]