Update autotrack.py

autotrack.py

@@ -1,19 +1,61 @@
 # 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
+# Auto-track milestones (e.g., 2×) for any Contract Address (CA) mentioned in NEW messages
+# from YOUR group only: https://t.me/MidasTouchsignalll
+#
+# Data: Dexscreener (free) + fallback Jupiter (Solana)
+# Deps: telethon, aiohttp
 
 import os
+import re
 import time
 import math
 import asyncio
 import aiohttp
 from dataclasses import dataclass, field
 from typing import Optional, Dict, List, Tuple
+from datetime import datetime, timezone, timedelta
 
+from telethon import TelegramClient, events
+from telethon.sessions import StringSession, MemorySession
+from telethon.errors.rpcerrorlist import FloodWaitError
+
+# =========================
+# ENV / CONFIG
+# =========================
+API_ID = int(os.environ.get("API_ID", "0"))
+API_HASH = os.environ.get("API_HASH", "")
+STRING_SESSION = os.environ.get("STRING_SESSION", "")
+
+# Announce ke grup kamu (dipakai juga sebagai sumber pesan)
+TARGET_CHAT = os.environ.get("TARGET_CHAT", "https://t.me/MidasTouchsignalll")
+
+# Polling interval harga
+TRACK_POLL_SECS = int(os.environ.get("TRACK_POLL_SECS", "20"))
+
+# Milestones "x" sebagai string koma, contoh: "1.5,2,3"
+_ms_env = os.environ.get("MILESTONES", "2")
+try:
+    DEFAULT_MILESTONES = sorted({float(x.strip()) for x in _ms_env.split(",") if x.strip()})
+except Exception:
+    DEFAULT_MILESTONES = [2.0]
+
+STOP_WHEN_HIT = True  # stop setelah milestone terbesar tercapai
+
+# Abaikan pesan lebih tua dari (startup - buffer)
+BACKFILL_BUFFER_MINUTES = int(os.environ.get("BACKFILL_BUFFER_MINUTES", "3"))
+
+# Marker antirecursive: ditambahkan ke semua pengumuman bot
+BOT_MARKER = "【MT-AUTOTRACK】"
+
+# =========================
+# HTTP endpoints
+# =========================
 DEXSCREENER_TOKEN_URL = "https://api.dexscreener.com/latest/dex/tokens/"
-JUPITER_URL = "https://price.jup.ag/v6/price?ids="  # gratis untuk Solana
+JUPITER_URL = "https://price.jup.ag/v6/price?ids="  # free (Solana only)
 
+# =========================
+# Helpers
+# =========================
 def _fmt_money(x: Optional[float]) -> str:
     if x is None:
         return "?"
@@ -24,7 +66,6 @@ def _fmt_money(x: Optional[float]) -> str:
 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:
@@ -33,15 +74,49 @@ def _fmt_big(x: Optional[float]) -> str:
         return f"${x/1_000:.2f}K"
     return f"${x:.0f}"
 
+# =========================
+# CA extraction
+# =========================
+CA_SOL_RE = re.compile(r"\b[1-9A-HJ-NP-Za-km-z]{32,48}\b")   # Solana base58 (approx)
+CA_EVM_RE = re.compile(r"\b0x[a-fA-F0-9]{40}\b")             # EVM 0x...
+
+CA_LABEL_RE = re.compile(r"\bCA\s*[:=]\s*(\S+)", re.IGNORECASE)
+
+def extract_ca(text: str) -> Optional[str]:
+    """Return first found CA (EVM or Solana). Prefer 'CA: <addr>' label if present."""
+    if not text:
+        return None
+
+    # Prefer explicit "CA: ..."
+    m = CA_LABEL_RE.search(text)
+    if m:
+        cand = m.group(1)
+        if CA_EVM_RE.fullmatch(cand):
+            return cand.lower()
+        if CA_SOL_RE.fullmatch(cand):
+            return cand
+
+    # Else, scan for EVM then Sol
+    m2 = CA_EVM_RE.search(text)
+    if m2:
+        return m2.group(0).lower()
+    m3 = CA_SOL_RE.search(text)
+    if m3:
+        return m3.group(0)
+    return None
+
+# =========================
+# Tracker core
+# =========================
 @dataclass
 class TrackItem:
     ca: str
-    basis: str = "mcap"  # "mcap" (default) atau "price"
+    basis: str = "mcap"  # "mcap" (default) or "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]
+    milestones: List[float] = field(default_factory=lambda: [2.0])
     hit: Dict[float, bool] = field(default_factory=dict)
     poll_secs: int = 20
     stop_when_hit: bool = True
@@ -49,27 +124,17 @@ class TrackItem:
 
 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
-        )
+    Usage:
+        tracker = PriceTracker(client, announce_chat=TARGET_CHAT)
+        await tracker.start(ca=..., basis="mcap", milestones=[2], poll_secs=20, source_link=..., symbol_hint=...)
     """
-    def __init__(self, client, announce_chat):
+    def __init__(self, client: TelegramClient, announce_chat):
         self.client = client
         self.announce_chat = announce_chat
         self._tasks: Dict[str, asyncio.Task] = {}
         self._items: Dict[str, TrackItem] = {}
 
-    # =========================
-    # HTTP helpers
-    # =========================
+    # ------------- 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:
@@ -80,8 +145,8 @@ class PriceTracker:
 
     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.
+        Return (priceUsd, symbol, marketCapUsd/FDV) from 'best pair' by highest USD liquidity.
+        price or mcap may be None if unavailable.
         """
         data = await self._get(DEXSCREENER_TOKEN_URL + ca)
         if not data:
@@ -89,27 +154,21 @@ class PriceTracker:
         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).
-        """
+        """Fallback for Solana via Jupiter."""
         try:
             data = await self._get(JUPITER_URL + ca)
             if not data:
@@ -123,47 +182,71 @@ class PriceTracker:
             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).
-        """
+        """Return dict {price, mcap, symbol_hint} from Dexscreener then Jupiter fallback."""
         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
-    # =========================
+    # ------------- Announce helpers (brand EN) -------------
     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)
+            await self.client.send_message(self.announce_chat, f"{text}\n\n{BOT_MARKER}", link_preview=False)
         except Exception as e:
             print(f"[TRACK] announce failed: {e}")
 
-    # =========================
-    # Core loop
-    # =========================
+    def _start_text(self, item: TrackItem, basis_label: str) -> str:
+        entry_disp = _fmt_big(item.entry_mcap) if item.basis == "mcap" else _fmt_money(item.entry_price)
+        return (
+            "💎 [MidasTouch Signal] 💎\n"
+            "New contract detected:\n\n"
+            f"CA: `{item.ca}`\n\n"
+            f"Tracking Basis: **{basis_label}**\n"
+            f"Entry: {entry_disp} • Milestones: {', '.join([f'{m}×' for m in item.milestones])}\n"
+            + (f"🔗 {item.source_link}\n" if item.source_link else "")
+            + "\nRemember: This is a signal, not financial advice.\n"
+            "Stay safe, stay golden ✨"
+        )
+
+    def _milestone_text(self, item: TrackItem, m: float, ratio: float, cur_price: Optional[float], cur_mcap: Optional[float]) -> str:
+        if item.basis == "mcap":
+            change = f"{_fmt_big(item.entry_mcap)} → {_fmt_big(cur_mcap)}  (~{ratio:.2f}×)"
+            milestone_line = f"Milestone reached: **{m}× Market Cap**"
+        else:
+            change = f"{_fmt_money(item.entry_price)} → {_fmt_money(cur_price)}  (~{ratio:.2f}×)"
+            milestone_line = f"Milestone reached: **{m}× Price**"
+
+        elapsed = int(time.time() - item.started_at)
+        mm, ss = divmod(elapsed, 60)
+        return (
+            "💎 [MidasTouch Signal] 💎\n"
+            f"{milestone_line}\n"
+            f"{self._name(item)}\n"
+            f"{change}\n"
+            f"⏱️ {mm}m {ss}s since call\n"
+            + (f"🔗 {item.source_link}\n" if item.source_link else "")
+            + f"CA: `{item.ca}`\n\n"
+            "Remember: This is a signal, not financial advice.\n"
+            "Stay safe, stay golden ✨"
+        )
+
+    # ------------- Loop -------------
     async def _loop(self, item: TrackItem):
-        # Inisialisasi entry dari snapshot
+        # init 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).")
+            await self._say(f"⚠️ Failed to start tracking for `{item.ca}` (no initial price/mcap).")
             return
 
         if item.entry_price is None:
@@ -173,21 +256,12 @@ class PriceTracker:
         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}`"
-        )
+        await self._say(self._start_text(item, basis_label))
 
-        # Loop polling
         while True:
             snap = await self._get_snapshot(item.ca)
             if not snap:
@@ -199,7 +273,7 @@ class PriceTracker:
             if not item.symbol_hint and snap.get("symbol_hint"):
                 item.symbol_hint = snap.get("symbol_hint")
 
-            # hitung rasio sesuai basis
+            # ratio
             if item.basis == "mcap":
                 if not (item.entry_mcap and cur_mcap):
                     await asyncio.sleep(item.poll_secs)
@@ -211,43 +285,20 @@ class PriceTracker:
                     continue
                 ratio = (cur_price / item.entry_price) if item.entry_price > 0 else 0.0
 
-            # cek milestones
+            # 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)
-
+                    await self._say(self._milestone_text(item, m, ratio, cur_price, cur_mcap))
                     if item.stop_when_hit and m == max(item.milestones):
-                        await self._say(f"🛑 Tracking dihentikan untuk {self._name(item)} (milestone akhir tercapai).")
+                        await self._say(f"🛑 Tracking stopped for {self._name(item)} (final milestone reached).")
                         return
 
             await asyncio.sleep(item.poll_secs)
 
-    # =========================
-    # Public API
-    # =========================
+    # ------------- Public -------------
     def is_tracking(self, ca: str) -> bool:
         t = self._tasks.get(ca)
         return bool(t) and not t.done()
@@ -259,7 +310,7 @@ class PriceTracker:
         basis: str = "mcap",
         entry_price: Optional[float] = None,
         entry_mcap: Optional[float] = None,
-        milestones: List[float] = [2.0],
+        milestones: List[float] = None,
         poll_secs: int = 20,
         source_link: Optional[str] = None,
         symbol_hint: Optional[str] = None,
@@ -268,6 +319,9 @@ class PriceTracker:
         ca = ca.strip()
         if self.is_tracking(ca):
             return
+        if milestones is None or not milestones:
+            milestones = DEFAULT_MILESTONES
+
         item = TrackItem(
             ca=ca,
             basis=basis.lower(),
@@ -279,15 +333,132 @@ class PriceTracker:
             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()
+
+# =========================
+# Setup: attach to existing client from botsignal
+# =========================
+def setup_autotrack(shared_client: TelegramClient, announce_chat: str | None = None):
+    """
+    Attach autotrack to an existing Telethon client.
+    - Registers the new-message handler for TARGET_CHAT
+    - Initializes global tracker with the shared client
+    """
+    global client, tracker, startup_time_utc
+    client = shared_client
+    if announce_chat:
+        ac = announce_chat
+    else:
+        ac = TARGET_CHAT
+    # initialize tracker with shared client
+    tracker = PriceTracker(client, announce_chat=ac)
+    # startup timestamp for anti-backfill
+    startup_time_utc = datetime.now(timezone.utc)
+    # register handler
+    client.add_event_handler(on_new_message, events.NewMessage(chats=(TARGET_CHAT,)))
+    print("[AUTOTRACK] attached to shared client; listening on", TARGET_CHAT)
+
+# =========================
+# Client bootstrap
+# =========================
+def build_client() -> TelegramClient:
+    if STRING_SESSION:
+        print(">> Using StringSession (persistent).")
+        return TelegramClient(StringSession(STRING_SESSION), API_ID, API_HASH)
+    print(">> Using MemorySession (login each run).")
+    return TelegramClient(MemorySession(), API_ID, API_HASH)
+
+client: TelegramClient | None = None
+startup_time_utc = None
+me_user_id: Optional[int] = None  # filled at start
+
+tracker: PriceTracker | None = None
+
+# =========================
+# Event handler: ONLY your group
+# =========================
+CUTOFF_BUFFER = timedelta(minutes=BACKFILL_BUFFER_MINUTES)
+
+def _is_old_message(msg_dt: Optional[datetime]) -> bool:
+    if not isinstance(msg_dt, datetime):
+        return False
+    # Telethon new-message handler biasanya sudah hanya message baru.
+    # Buffer ini sebagai pengaman tambahan.
+    return msg_dt.replace(tzinfo=timezone.utc) < (startup_time_utc - CUTOFF_BUFFER)
+
+def _is_bot_own_message(event) -> bool:
+    # Jangan proses pesan yang dikirim oleh diri sendiri (script ini)
+    if getattr(event, "out", False):
+        return True
+    txt = (event.raw_text or "") if hasattr(event, "raw_text") else ""
+    return BOT_MARKER in (txt or "")
+
+async def on_new_message(event):
+    try:
+        # anti-loop: jangan proses pesan sendiri/marker
+        if _is_bot_own_message(event):
+            return
+
+        msg = event.message
+        if _is_old_message(getattr(msg, "date", None)):
+            # ignore any old message near startup (extra safety)
+            return
+
+        text = msg.message or (getattr(msg, "raw_text", None) or "")
+        ca = extract_ca(text)
+        if not ca:
+            return  # no CA found
+
+        # Optional: link back to source message
+        source_link = None
+        try:
+            # If chat has a public username, build t.me link
+            chat = await event.get_chat()
+            uname = getattr(chat, "username", None)
+            mid = getattr(msg, "id", None)
+            if uname and mid:
+                source_link = f"https://t.me/{uname}/{mid}"
+        except:
+            pass
+
+        # Start tracking
+        await tracker.start(
+            ca=ca,
+            basis="mcap",
+            milestones=DEFAULT_MILESTONES,
+            poll_secs=TRACK_POLL_SECS,
+            source_link=source_link,
+            stop_when_hit=STOP_WHEN_HIT,
+        )
+        # (Optional) send small ack? Usually start_text already covers announce.
+        # Here we rely on _start_text inside tracker._loop
+    except Exception as e:
+        print(f"[AUTOTRACK] error: {e}")
+
+# =========================
+# Entrypoint
+# =========================
+
+async def main():
+    global client, tracker, startup_time_utc
+    if client is None:
+        client = build_client()
+    startup_time_utc = datetime.now(timezone.utc)
+    tracker = PriceTracker(client, announce_chat=TARGET_CHAT)
+    await client.start()
+    try:
+        me = await client.get_me()
+        globals()["me_user_id"] = int(getattr(me, "id", 0))
+        print(f">> Logged in as: {getattr(me, 'username', None) or me_user_id}")
+    except Exception as e:
+        print(f"Warning: cannot resolve self id: {e}")
+
+    print("AutoTrack running. Listening ONLY your group for NEW messages...")
+    # ensure handler is attached in standalone mode
+    client.add_event_handler(on_new_message, events.NewMessage(chats=(TARGET_CHAT,)))
+    await client.run_until_disconnected()
+
+if __name__ == "__main__":
+    asyncio.run(main())