Update autotrack.py

autotrack.py

@@ -0,0 +1,293 @@
+# autotrack.py
+# Auto-tracking milestone (mis. x2) dari ENTRY (price atau market cap) untuk sebuah CA.
+# Sumber data: Dexscreener (gratis, tanpa API key) + fallback Jupiter (Solana).
+# Dependensi: aiohttp
+
+import os
+import time
+import math
+import asyncio
+import aiohttp
+from dataclasses import dataclass, field
+from typing import Optional, Dict, List, Tuple
+
+DEXSCREENER_TOKEN_URL = "https://api.dexscreener.com/latest/dex/tokens/"
+JUPITER_URL = "https://price.jup.ag/v6/price?ids="  # gratis untuk Solana
+
+def _fmt_money(x: Optional[float]) -> str:
+    if x is None:
+        return "?"
+    if x >= 1:
+        return f"${x:,.4f}"
+    return f"${x:.8f}"
+
+def _fmt_big(x: Optional[float]) -> str:
+    if x is None:
+        return "?"
+    # Format untuk market cap
+    if x >= 1_000_000_000:
+        return f"${x/1_000_000_000:.2f}B"
+    if x >= 1_000_000:
+        return f"${x/1_000_000:.2f}M"
+    if x >= 1_000:
+        return f"${x/1_000:.2f}K"
+    return f"${x:.0f}"
+
+@dataclass
+class TrackItem:
+    ca: str
+    basis: str = "mcap"  # "mcap" (default) atau "price"
+    entry_price: Optional[float] = None
+    entry_mcap: Optional[float] = None
+    symbol_hint: Optional[str] = None
+    source_link: Optional[str] = None
+    milestones: List[float] = field(default_factory=lambda: [2.0])  # contoh: [1.5, 2.0, 3.0]
+    hit: Dict[float, bool] = field(default_factory=dict)
+    poll_secs: int = 20
+    stop_when_hit: bool = True
+    started_at: float = field(default_factory=time.time)
+
+class PriceTracker:
+    """
+    Cara pakai (contoh):
+        tracker = PriceTracker(client, announce_chat="@MidasTouchSignall")
+        await tracker.start(
+            ca="GXKC9BCWCiYVeEpQ8cFoFHpgAeXTzX3YCFX4BnaZpump",
+            basis="mcap",                     # atau "price"
+            milestones=[2.0],                 # x2
+            poll_secs=20,
+            source_link=None,                 # opsional link ke pesan sumber
+            symbol_hint="COIN",               # opsional
+            stop_when_hit=True
+        )
+    """
+    def __init__(self, client, announce_chat):
+        self.client = client
+        self.announce_chat = announce_chat
+        self._tasks: Dict[str, asyncio.Task] = {}
+        self._items: Dict[str, TrackItem] = {}
+
+    # =========================
+    # HTTP helpers
+    # =========================
+    async def _get(self, url: str, headers: Optional[Dict[str, str]] = None, timeout: int = 8):
+        tout = aiohttp.ClientTimeout(total=timeout)
+        async with aiohttp.ClientSession(timeout=tout, headers=headers) as sess:
+            async with sess.get(url) as r:
+                if r.status != 200:
+                    return None
+                return await r.json()
+
+    async def _dexscreener_price(self, ca: str) -> Optional[Tuple[Optional[float], Optional[str], Optional[float]]]:
+        """
+        Mengembalikan (priceUsd, symbol, marketCapUsd/FDV) dari pair 'terbaik' (likuiditas USD terbesar).
+        Bisa saja price atau mcap None bila tidak tersedia.
+        """
+        data = await self._get(DEXSCREENER_TOKEN_URL + ca)
+        if not data:
+            return None
+        pairs = data.get("pairs") or []
+        if not pairs:
+            return None
+
+        # Pilih pair dengan likuiditas USD terbesar (heuristik stabil)
+        best = max(pairs, key=lambda p: (p.get("liquidity", {}) or {}).get("usd", 0))
+        price = float(best.get("priceUsd")) if best.get("priceUsd") else None
+        symbol = ((best.get("baseToken") or {}).get("symbol")) or None
+
+        mcap = None
+        # Dexscreener kadang expose 'fdv' &/atau 'marketCap' pada pair
+        fdv = best.get("fdv")
+        if isinstance(fdv, (int, float)) and fdv > 0:
+            mcap = float(fdv)
+        mc = best.get("marketCap")
+        if isinstance(mc, (int, float)) and mc > 0:
+            mcap = float(mc)
+
+        return (price, symbol, mcap)
+
+    async def _jupiter_price(self, ca: str) -> Optional[float]:
+        """
+        Fallback harga khusus Solana via Jupiter (gratis).
+        """
+        try:
+            data = await self._get(JUPITER_URL + ca)
+            if not data:
+                return None
+            d = (data.get("data") or {}).get(ca)
+            if not d:
+                return None
+            p = d.get("price")
+            return float(p) if p is not None else None
+        except:
+            return None
+
+    async def _get_snapshot(self, ca: str) -> Optional[Dict[str, Optional[float]]]:
+        """
+        Ambil snapshot {price, mcap, symbol_hint}.
+        Urutan fallback: Dexscreener -> Jupiter (Solana only).
+        """
+        res = await self._dexscreener_price(ca)
+        if res:
+            price, sym, mcap = res
+            # kalau price kosong, coba fallback Jupiter
+            if price is None:
+                jp = await self._jupiter_price(ca)
+                price = jp if jp is not None else None
+            return {"price": price, "symbol_hint": sym, "mcap": mcap}
+
+        # kalau Dexscreener gagal total → coba Jupiter (harga saja)
+        jp = await self._jupiter_price(ca)
+        if jp is not None:
+            return {"price": jp, "symbol_hint": None, "mcap": None}
+
+        return None
+
+    # =========================
+    # Helpers UI
+    # =========================
+    def _name(self, item: TrackItem) -> str:
+        s = item.symbol_hint or ""
+        return f"{s} ({item.ca[:4]}…{item.ca[-4:]})" if s else item.ca
+
+    async def _say(self, text: str):
+        try:
+            await self.client.send_message(self.announce_chat, text, link_preview=False)
+        except Exception as e:
+            print(f"[TRACK] announce failed: {e}")
+
+    # =========================
+    # Core loop
+    # =========================
+    async def _loop(self, item: TrackItem):
+        # Inisialisasi entry dari snapshot
+        snap = await self._get_snapshot(item.ca)
+        if not snap:
+            await self._say(f"⚠️ Gagal mulai tracking untuk `{item.ca}` (tidak dapat harga/mcap awal).")
+            return
+
+        if item.entry_price is None:
+            item.entry_price = snap.get("price")
+        if item.entry_mcap is None:
+            item.entry_mcap = snap.get("mcap")
+        if not item.symbol_hint and snap.get("symbol_hint"):
+            item.symbol_hint = snap.get("symbol_hint")
+
+        # Jika basis=mcap tapi mcap awal tidak ada → fallback ke basis=price
+        if item.basis == "mcap" and not item.entry_mcap:
+            item.basis = "price"
+
+        basis_label = "Market Cap" if item.basis == "mcap" else "Price"
+        entry_disp = _fmt_big(item.entry_mcap) if item.basis == "mcap" else _fmt_money(item.entry_price)
+
+        await self._say(
+            f"🛰️ Tracking: **{self._name(item)}**  | Basis: **{basis_label}**\n"
+            f"Entry: {entry_disp} • Milestones: {', '.join([f'{m}×' for m in item.milestones])}"
+            + (f"\n🔗 {item.source_link}" if item.source_link else "")
+            + f"\nCA: `{item.ca}`"
+        )
+
+        # Loop polling
+        while True:
+            snap = await self._get_snapshot(item.ca)
+            if not snap:
+                await asyncio.sleep(item.poll_secs)
+                continue
+
+            cur_price = snap.get("price")
+            cur_mcap = snap.get("mcap")
+            if not item.symbol_hint and snap.get("symbol_hint"):
+                item.symbol_hint = snap.get("symbol_hint")
+
+            # hitung rasio sesuai basis
+            if item.basis == "mcap":
+                if not (item.entry_mcap and cur_mcap):
+                    await asyncio.sleep(item.poll_secs)
+                    continue
+                ratio = (cur_mcap / item.entry_mcap) if item.entry_mcap > 0 else 0.0
+            else:
+                if not (item.entry_price and cur_price):
+                    await asyncio.sleep(item.poll_secs)
+                    continue
+                ratio = (cur_price / item.entry_price) if item.entry_price > 0 else 0.0
+
+            # cek milestones
+            for m in item.milestones:
+                if item.hit.get(m):
+                    continue
+                if ratio >= m:
+                    item.hit[m] = True
+                    elapsed = int(time.time() - item.started_at)
+                    mm, ss = divmod(elapsed, 60)
+                    if item.basis == "mcap":
+                        msg = (
+                            "✅ **AUTO-TRACK**\n"
+                            f"{self._name(item)} menembus **{m}×** mcap\n"
+                            f"{_fmt_big(item.entry_mcap)} → {_fmt_big(cur_mcap)}  (~{ratio:.2f}×)\n"
+                            f"⏱️ {mm}m {ss}s sejak call"
+                            + (f"\n🔗 {item.source_link}" if item.source_link else "")
+                            + f"\nCA: `{item.ca}`"
+                        )
+                    else:
+                        msg = (
+                            "✅ **AUTO-TRACK**\n"
+                            f"{self._name(item)} menembus **{m}×** harga\n"
+                            f"{_fmt_money(item.entry_price)} → {_fmt_money(cur_price)}  (~{ratio:.2f}×)\n"
+                            f"⏱️ {mm}m {ss}s sejak call"
+                            + (f"\n🔗 {item.source_link}" if item.source_link else "")
+                            + f"\nCA: `{item.ca}`"
+                        )
+                    await self._say(msg)
+
+                    if item.stop_when_hit and m == max(item.milestones):
+                        await self._say(f"🛑 Tracking dihentikan untuk {self._name(item)} (milestone akhir tercapai).")
+                        return
+
+            await asyncio.sleep(item.poll_secs)
+
+    # =========================
+    # Public API
+    # =========================
+    def is_tracking(self, ca: str) -> bool:
+        t = self._tasks.get(ca)
+        return bool(t) and not t.done()
+
+    async def start(
+        self,
+        ca: str,
+        *,
+        basis: str = "mcap",
+        entry_price: Optional[float] = None,
+        entry_mcap: Optional[float] = None,
+        milestones: List[float] = [2.0],
+        poll_secs: int = 20,
+        source_link: Optional[str] = None,
+        symbol_hint: Optional[str] = None,
+        stop_when_hit: bool = True,
+    ):
+        ca = ca.strip()
+        if self.is_tracking(ca):
+            return
+        item = TrackItem(
+            ca=ca,
+            basis=basis.lower(),
+            entry_price=entry_price,
+            entry_mcap=entry_mcap,
+            milestones=sorted(set(milestones)),
+            poll_secs=poll_secs,
+            source_link=source_link,
+            symbol_hint=symbol_hint,
+            stop_when_hit=stop_when_hit,
+        )
+        self._items[ca] = item
+        self._tasks[ca] = asyncio.create_task(self._loop(item))
+
+    def stop(self, ca: str):
+        t = self._tasks.get(ca)
+        if t and not t.done():
+            t.cancel()
+
+    def stop_all(self):
+        for t in list(self._tasks.values()):
+            if t and not t.done():
+                t.cancel()