aetheris_memory.py

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]