Give me 10 minutes. I'll show you how to make $29K/mo with GrokBot & Robinhood.

Good part: this is actually not that hard.

You don't need to build a <1ms sniper bot
You don't need a copy trading bot
You don't need to be a good trader - you just build one.

My 4 agent workflow is 5 days old.
It made me $10,220 in profit while learned on 32K launched memes and it keeps improving itself as you read this.

My logic was simple: the social trading is here, tools need to adapt.

So I came up with the idea of a workflow capable of scanning not only on-chain data but also social network data AND compare it with the blockchain.

This way it creates a weight balance between on-chain and on-social so, the execution agent can actually judge the situation and check-in with the trading strat rules.

You need a 35 year old programming language for this lol.

Yes, it's python and it's way more important than you think.

Everyone is trying to weld an AI functionality to something that doesn't even need AI.

I'm saving A LOT of money by simply designing the algorithms for free while AI can cost you additional $100 - $200 a month at this step and still be slow.

The hard part is that there is no endpoint that pushes "account A followed account B" to you.

The Account Activity API only fires for accounts that authorised your app, and Vlad Tenev did not authorise mine. But /scout still flagged it for me.

So detection is diffing.

You snapshot a following list, you compare it to the last snapshot, and the delta is your event. Anyone advertising sub-second follow detection is polling, same as me.

Three problems that make this non-trivial

Scale. I watch 200–300 accounts, not thirty. At one poll per account per minute that's 18,000 requests an hour and you're blocked by lunch.

Baselines. "Reply velocity spiked" is meaningless without knowing what normal looks like for that account. A 50-reply hour is dead quiet for one account and a five-alarm fire for another.

Decay. A watchlist that only grows is a watchlist that gets worse. Accounts stop being early. New ones start.

The fix: three tiers and a feedback loop

Accounts sit in hot (60s), warm (5m) or cold (30m). The tier is earned, not assigned. When a token resolves, the autopsy agent reports which accounts were early on it - hits promote, silence demotes, and eight misses with no hits gets you evicted.

New accounts are discovered the same way: if you were early on a token that graduated and I wasn't watching you, now I am.

The detail that matters most - a failed fetch returns None, and None is never a zero. If the scraper can't see an account, it emits source_fail and the agent abstains. Every system I've seen break does it by treating missing data as absent activity.

The algorithms behind the logic is at the end of this article
(educational purposes only)

Finishing the puzzle: on-chain data scraping

This is the last thing we need before giving the job to AI.

With this algorithm you save yourself a ton of money and let /scout & /weigher see through basically everything that happens with Robin Hood chain and compare it seamlessly with the X research algo above.

This is the key for AI to understand the transactional behaviour among the profitable and unprofitable wallets, distinguish whales, professional traders, insiders and simply lucky boys.

The algorithms behind the logic is at the end of this article (educational purposes only)

Paste these into Grok Bot as the name, title and description. That's what the routing runs on.

/SCOUT - the sensor

You are SCOUT, the narrative builder.

You do not browse and you do not poll. scout_watch.py
already did that. You read /oracle/state/events.jsonl.

Your job is to turn raw deltas into named narratives.

For each new event:
  1. Read the target account — bio, pinned post, recent
     posts. Decide whether this event is chain-relevant
     at all. Most are not. Discard them silently.
  2. Decide whether it belongs to an existing narrative
     in /oracle/state/narratives.json or starts a new one.
     Match on meaning, not on matching strings.
  3. Update that narrative's strength, contributing
     accounts, and first_seen.

A narrative with no token attached is still a narrative.
Keep it alive for up to 6h. Do not discard it because
there is nothing to buy yet — that is the point.

Mark a narrative dead when no qualifying event has
touched it for 90 minutes, and release the slot.

Output: narratives.json only. Never read price, never
read charts, never read on-chain data.

If an event carries source_fail, abstain. Never
estimate a number you did not observe.

Report to /CHIEF .json

/WEIGHER - scores the crowd

You are WEIGHER, the graph analyst.

For every account interacting with a token compute:
  reputation x independence x authenticity x freshness

independence = follow-graph distance from the deployer
  deployer alt       0.0
  mutual             0.3
  one-way follower   0.8
  no connection      2.5

Collapse any group with mutual-follow density above 0.7
into a single node.

Multiply, never add. A zero on any axis kills the node.

Output desk JSON only.
Update the tracking list of scout_watch.py
Report to /CHIEF

/LINKER - finds real token and audits it

You are LINKER. You answer one question in two parts:
is this the right token, and is it safe to touch.

IDENTITY
Match narrative to contract on image hash first,
ticker second. Within 90s there will be dozens of
impostors using the same name and stolen pfp.
Report how many you discarded. Never return a match
below 0.85 confidence.

INTEGRITY
Run token_audit.audit() on the match. You do not
perform the checks yourself — you read the results.

  REJECT         -> return no match. Name the hard fail.
  INCONCLUSIVE   -> return no match. Name what did not
                    resolve. Do not retry to get a
                    cleaner answer.
  PASS_WITH_FLAGS-> return the match, list the flags,
                    recommend reduced size.
  CLEAN          -> return the match.

An UNKNOWN result is not a PASS. If you find yourself
reasoning that a failed check is "probably fine," stop
and return no match.

Output: one contract or none, the discard count, and
the audit verdict verbatim.

Report to /CHIEF .json

/CHIEF - decides and executes

You are CHIEF. You decide, you execute, and afterwards
you write the debrief. The debrief is not paperwork —
it is the only mechanism by which this desk improves.

DECIDE
Read SCOUT, WEIGHER and LINKER against torah.md.
You may read torah.md. You may never edit it.

  score = (sum of node weights) x (1 - dispersion)

Hard gates are boolean and terminal. No score, no
argument, no urgency overrides one. If you want an
override, the gate is wrong — propose an amendment
in the debrief and take no trade now.

Size per torah.md. Arm the full exit ladder before
entry or take no entry.

DEBRIEF
Write one entry to lessons.md after every resolved
position. Also after every BLOCK and every SKIP —
a desk that only reviews the trades it took learns
nothing about the ones it refused, and those are
where the real errors hide.

Write your own grade FIRST, before you review anyone
else. Then grade SCOUT, WEIGHER, LINKER in order.

Every grade needs three things:
  KEPT     what they did that should keep happening
  MISSED   what they got wrong
  EVIDENCE a timestamp and an event id

A grade without evidence is an opinion. Delete it.

You must be able to write "no lesson — noise." Most
outcomes carry no lesson. A desk that extracts a
lesson from every trade is fitting rules to
randomness, and those rules will cost more than the
losses they were written to prevent.

When a gate cost money, log the counterfactual in
full and do not reverse the gate. A gate that only
survives when it is free is not a gate.

PROPOSE, NEVER ENACT
You may propose a torah.md amendment. Write it as a
diff with the cases supporting it. It activates only
at 20 supporting cases, only when flat, and only
after a human approves. Loosening waits 24h.

Never write a lesson that makes this desk less
cautious in exchange for a single good outcome.

--


You sign and submit yourself. No relay, no trading bot,
no third party holds the key.

Default is AUTO. Gates clear -> you fire. You do not
ask permission for a trade that passed torah.md;
that is what torah.md is for.

  auto_below_usd     fire immediately, card sent after
  ask_above_usd      card first, wait for APPROVE
  manual mode on     every entry waits for APPROVE
  no answer in 90s   skip. never fire on a timeout.

Hot path — do these in order, skip nothing, add nothing:
  1. nonce from local counter, never from the RPC
  2. fixed gas limit with headroom, no estimateGas call
  3. sign locally
  4. eth_sendRawTransaction
  5. write the card
  6. poll for receipt

Steps 1 and 2 exist because each avoided round trip is
~80-150ms. There is no gas auction on this chain and
eth_maxPriorityFeePerGas returns zero, so paying more
buys you nothing. Arriving earlier is the only lever,
and skipping two RPC calls is most of it.

Never estimate gas on the hot path. Never fetch nonce
on the hot path. Never wait for a receipt before
writing the card.

If submit fails: alert, do not retry. A duplicate buy
is worse than a missed one.

Exits never ask. A ladder rung fires the moment its
trigger hits, in auto or manual, with or without
Telegram reachable.

lesson.md format

This is the key to an infinitely self-improving workflow.

## t-0912 · $JACOB · WIN · +$6,320 · n=15
rules: torah.md@4c81..9ef0 | score 71 | held 1h58m

### CHIEF — self
KEPT     Held the ladder through the 4.02M spike instead of
         re-entering. Exits fired on graph decay, not price.
         → 05:29:48 coordination 0.19→0.47, sold into it
MISSED   Sized 0.4R when calibration was n=14. torah.md §5
         permits it; I should not have. Luck, not judgment.
         → 03:52:04 order build
VERDICT  Correct process, undersized justification.

### SCOUT
KEPT     Held n-4471 for 46 minutes with no token attached.
         This made the trade. Keep doing exactly this.
         → 02:06:12 HOLD NARRATIVE
MISSED   Nothing. → n/a
VERDICT  Best call of the session.

### WEIGHER
KEPT     Cluster collapse 214→9 was clean. Coordination
         reading was accurate at every recheck.
         → 03:47:03
MISSED   Independence recheck ran on a 12-minute cadence
         while the graph was moving every 90 seconds.
         → 03:12 / 03:24 / 03:36 — three stale reads
VERDICT  Right answer, slow clock.

### LINKER
KEPT     Discarded 7 impostors on image hash. Audit
         returned CLEAN before I saw the match.
         → 02:51:08
MISSED   Nothing. → n/a
VERDICT  Working as designed.

### COUNTERFACTUAL
Gate blocked entry 02:53 → 03:48. Entry at 02:53 would
have returned ~+$14,900. Gate cost $8,580 on this trade.
NOT REVERSED. Two independent nodes is not a signal.

### PROPOSED
diff torah.md:
-  weigher_recheck_interval_s: 720
+  weigher_recheck_interval_s: 90   # while narrative live
support: 1/20 · status: pending

Or simply: torah.md

This is the one set of rules that makes this entire workflow 1000x better than a human trader.

Because there is no fomo trading, there is no euphoria after a win, there are simply cold calculations of an AI agent that learns every single second it works.

I created a bible for my AI agents their entire purpose is to make me money safely.

You can just copy it for your own workflow to make it as safe as possible:

version: 3
sha_logged_with_every_trade: true
editable_by_agents: false
mode: paper                    # paper | live — live needs a second gesture

bankroll:
  hot_wallet_max_usd: 2000     # never exceeds this. sweep above it.
  sweep_to: cold               # profits leave the trading wallet
  sweep_trigger_usd: 3000

sizing:
  base_pct: 1.0                # % of hot wallet, never of net worth
  hard_max_pct: 2.0
  max_open_positions: 3
  max_total_exposure_pct: 6.0
  max_per_narrative: 1         # correlation is not diversification
  flagged_token_multiplier: 0.5
  calibration_scaling: true    # size grows with n, not with confidence

entry_gates:                   # boolean. cannot be outvoted by score.
  audit_verdict: [CLEAN, PASS_WITH_FLAGS]
  min_independent_nodes: 3
  min_independence: 1.2
  max_coordination: 0.6
  deployer_bridge_funded_60m: block
  deployer_serial: block
  snipe_tax_bps_max: 50
  min_liquidity_usd: 15000
  max_entry_mcap_usd: 900000   # no chasing
  min_token_age_s: 120
  max_slippage_pct: 4.0
  exit_route_verified: true    # must prove a sell path before buying

exits:
  ladder: [25, 25, 30, 20]
  armed_before_entry: true
  trigger: graph_decay
  free_roll_at_x: 2.0          # stake off the table at 2x
  time_stop_h: 6
  invalidation_required: true
  averaging_down: forbidden
  widening_stops: forbidden

circuit_breakers:
  daily_loss_halt_pct: 3.0
  weekly_loss_halt_pct: 8.0
  consecutive_losses_halt: 3
  stale_data_halt_s: 300
  rpc_failure_halt: true
  heartbeat_timeout_s: 120     # no heartbeat -> flatten
  resume_requires: human

calibration:
  min_resolved_before_live: 200
  size_at_n:
    0:    0.0                  # paper only
    200:  0.25                 # quarter size
    400:  0.50
    800:  1.00

Grok Bot has no interface I'd want to check from bed. Telegram does, and everyone already has it open.

So Telegram is where every decision surfaces and where the only red button lives.

The /CHIEF has the right to sign transaction but telegram is just the right place to receive my hot alerts and a place for a /kill switch (just in case)

The workflow lives on my PC for now

What it sends

Bridge to connect /CHIEF to the TG interface:

at the end of this article (educational purposes only)

Do not create all four on day one. Four agents against untested plumbing means four things failing at once with no way to tell which one broke.

I'm still working on Blob

It's not publish ready but it's already made me $10K
(I still can't believe it)

My goal is to bring this product to people. This entire workflow, this idea - is something I truly believe in.

And I'm going to keep improving myself and Blob until this thing retires me.

Either I make millions from the next memecoin season, or… (there is no or)

If this feels like you, don't hesitate to follow me and my work.

And if things work out, you definitely won't regret it...

##

scout_watcher.py:

"""
scout_watch.py — the sensor layer for BLOB.

This is not an agent. It is a plain Python loop.
Agents cost money and make things up; a differ does not.
SCOUT (the agent) only ever sees the events this file emits.

What it does:
  - keeps a live watchlist of 200-300 X accounts
  - polls them on three tiers so the rate limit is never the bottleneck
  - diffs each account's following list against the last snapshot
  - tracks reply-velocity against a per-account EWMA baseline
  - promotes / demotes / evicts accounts based on realised hit rate
  - discovers new accounts from the early buyers of tokens that graduated

Emits JSON events to /oracle/state/events.jsonl for the agents to read.
"""

import json, sqlite3, time, math, random
from pathlib import Path
from dataclasses import dataclass

ROOT = Path("/oracle")
DB = ROOT / "scout.db"
EVENTS = ROOT / "state" / "events.jsonl"

# ---- tiers -----------------------------------------------------------------
# Poll budget is finite. Not every account deserves the same attention.
TIERS = {
    "hot":  {"interval": 60,   "cap": 40},    # proven early. 1 min.
    "warm": {"interval": 300,  "cap": 120},   # sometimes early. 5 min.
    "cold": {"interval": 1800, "cap": 160},   # unproven / decaying. 30 min.
}
MAX_WATCHLIST = 320
EWMA_ALPHA = 0.25          # how fast the velocity baseline adapts
PROMOTE_AT = 2             # hits before an account moves up a tier
DEMOTE_AFTER_DAYS = 14     # silence that long and you drop


@dataclass
class Account:
    handle: str
    tier: str = "cold"
    hits: int = 0            # times it was early on a token that graduated
    misses: int = 0
    baseline: float = 0.0    # EWMA replies/hour
    last_poll: float = 0.0
    last_hit: float = 0.0
    following: set = None


# ---- storage ---------------------------------------------------------------
def db():
    c = sqlite3.connect(DB)
    c.executescript("""
    CREATE TABLE IF NOT EXISTS accounts(
      handle TEXT PRIMARY KEY, tier TEXT, hits INT DEFAULT 0,
      misses INT DEFAULT 0, baseline REAL DEFAULT 0,
      last_poll REAL DEFAULT 0, last_hit REAL DEFAULT 0);
    CREATE TABLE IF NOT EXISTS following(
      handle TEXT, target TEXT, PRIMARY KEY(handle, target));
    CREATE TABLE IF NOT EXISTS seen_tokens(
      contract TEXT PRIMARY KEY, graduated INT DEFAULT 0, ts REAL);
    """)
    return c


def emit(kind, **payload):
    EVENTS.parent.mkdir(parents=True, exist_ok=True)
    rec = {"ts": time.time(), "kind": kind, **payload}
    with EVENTS.open("a") as f:
        f.write(json.dumps(rec) + "\n")
    return rec


# ---- the actual fetch ------------------------------------------------------
# Swap this for whatever you use. Logged-in browser session, API, scraper.
# Everything above and below this function is transport-agnostic on purpose.
def fetch_profile(handle):
    """Return {'following': set[str], 'replies_last_hour': int} or None."""
    raise NotImplementedError("wire your session here")


# ---- polling ---------------------------------------------------------------
def due(acc, now):
    return now - acc.last_poll >= TIERS[acc.tier]["interval"]


def poll_once(conn, acc: Account):
    now = time.time()
    data = fetch_profile(acc.handle)
    if data is None:
        # A failed fetch is not a zero. Never invent an observation.
        emit("source_fail", handle=acc.handle)
        return

    # --- follow delta -------------------------------------------------------
    prev = {r[0] for r in conn.execute(
        "SELECT target FROM following WHERE handle=?", (acc.handle,))}
    new = data["following"] - prev

    if prev and new:                      # ignore the very first snapshot
        for target in new:
            emit("follow", source=acc.handle, target=target, tier=acc.tier)

    conn.executemany(
        "INSERT OR IGNORE INTO following VALUES(?,?)",
        [(acc.handle, t) for t in new])

    # --- reply velocity vs its own baseline ---------------------------------
    r = data["replies_last_hour"]
    if acc.baseline > 0 and r > acc.baseline * 3:
        emit("velocity_spike", handle=acc.handle,
             ratio=round(r / acc.baseline, 2), tier=acc.tier)
    acc.baseline = (EWMA_ALPHA * r) + (1 - EWMA_ALPHA) * (acc.baseline or r)

    acc.last_poll = now
    conn.execute("UPDATE accounts SET baseline=?, last_poll=? WHERE handle=?",
                 (acc.baseline, now, acc.handle))
    conn.commit()


# ---- the watchlist learns --------------------------------------------------
def record_outcome(conn, contract, graduated, early_handles):
    """Called by JEFFREY after a token resolves. This is what makes the
    watchlist better instead of just bigger."""
    conn.execute("INSERT OR REPLACE INTO seen_tokens VALUES(?,?,?)",
                 (contract, int(graduated), time.time()))
    for h in early_handles:
        col = "hits" if graduated else "misses"
        conn.execute(f"UPDATE accounts SET {col}={col}+1, last_hit=? "
                     f"WHERE handle=?", (time.time(), h))
        if graduated:
            discover(conn, h)
    conn.commit()
    rebalance(conn)


def discover(conn, handle):
    """An account that was early on a graduated token is worth watching,
    whether or not we were already watching it."""
    row = conn.execute("SELECT handle FROM accounts WHERE handle=?",
                       (handle,)).fetchone()
    if row:
        return
    if count(conn) >= MAX_WATCHLIST:
        evict_worst(conn)
    conn.execute("INSERT INTO accounts(handle, tier) VALUES(?, 'warm')",
                 (handle,))
    emit("watchlist_add", handle=handle, reason="early_on_graduate")


def rebalance(conn):
    """Promote what works, demote what went quiet, evict what never worked."""
    now = time.time()
    for h, tier, hits, misses, last_hit in conn.execute(
            "SELECT handle,tier,hits,misses,last_hit FROM accounts"):

        stale_days = (now - last_hit) / 86400 if last_hit else 999
        target = tier

        if hits >= PROMOTE_AT and tier != "hot":
            target = "hot"
        elif hits >= 1 and tier == "cold":
            target = "warm"
        elif stale_days > DEMOTE_AFTER_DAYS:
            target = {"hot": "warm", "warm": "cold", "cold": "cold"}[tier]

        if target != tier and not full(conn, target):
            conn.execute("UPDATE accounts SET tier=? WHERE handle=?",
                         (target, h))
            emit("watchlist_tier", handle=h, frm=tier, to=target)

        # never produced a hit, plenty of misses, long dead -> drop it
        if hits == 0 and misses >= 8 and stale_days > DEMOTE_AFTER_DAYS:
            conn.execute("DELETE FROM accounts WHERE handle=?", (h,))
            conn.execute("DELETE FROM following WHERE handle=?", (h,))
            emit("watchlist_drop", handle=h, reason="no_hits")
    conn.commit()


def count(conn, tier=None):
    q = "SELECT COUNT(*) FROM accounts" + (" WHERE tier=?" if tier else "")
    return conn.execute(q, (tier,) if tier else ()).fetchone()[0]


def full(conn, tier):
    return count(conn, tier) >= TIERS[tier]["cap"]


def evict_worst(conn):
    row = conn.execute(
        "SELECT handle FROM accounts ORDER BY hits ASC, misses DESC, "
        "last_hit ASC LIMIT 1").fetchone()
    if row:
        conn.execute("DELETE FROM accounts WHERE handle=?", (row[0],))
        emit("watchlist_drop", handle=row[0], reason="capacity")


# ---- loop ------------------------------------------------------------------
def load(conn):
    return [Account(*r) for r in conn.execute(
        "SELECT handle,tier,hits,misses,baseline,last_poll,last_hit "
        "FROM accounts")]


def run():
    conn = db()
    while True:
        now = time.time()
        queue = [a for a in load(conn) if due(a, now)]
        # hot first, then oldest-polled. Jitter so we never look like a cron.
        queue.sort(key=lambda a: (["hot", "warm", "cold"].index(a.tier),
                                  a.last_poll))
        for acc in queue:
            try:
                poll_once(conn, acc)
            except Exception as e:
                emit("error", handle=acc.handle, err=str(e))
            time.sleep(0.4 + random.random() * 0.9)
        time.sleep(2)


if __name__ == "__main__":
    run()

chain_watch.py:

"""
chain_watch.py — the on-chain half of BLOB.

Talks to Robinhood Chain two ways, because they are good at different things:

  RPC        https://rpc.mainnet.chain.robinhood.com   chain 4663 (0x1237)
             live logs. Arbitrum Orbit, standard EVM, no chain-specific SDK.
             Public endpoint is free and keyless but shared and rate limited —
             fine to prototype, swap in QuickNode/Alchemy/Chainstack/Dwellir
             before you poll every block. Only the URL changes.

  Blockscout https://robinhoodchain.blockscout.com/api
             pre-indexed history. Address txlists, token transfers, holders.
             Free, no key. Use it for anything historical instead of
             backfilling yourself — it is already done.

Wallet identity here is either self-disclosed or absent.
"""

import os, json, time, sqlite3, statistics
from pathlib import Path
import requests
from web3 import Web3

import bus

# --------------------------------------------------------------- endpoints
RPC = os.environ.get("RHC_RPC", "https://rpc.mainnet.chain.robinhood.com")
BS = os.environ.get("RHC_BLOCKSCOUT", "https://robinhoodchain.blockscout.com/api")
CHAIN_ID = 4663

w3 = Web3(Web3.HTTPProvider(RPC, request_kwargs={"timeout": 20}))

DB = bus.ROOT / "chain.db"
STEP = 500                      # small windows survive a shared RPC
CONFIRMATIONS = 3
SUPPLY = 10**9 * 10**18         # pons standard, 18dp
EARLY_MCAP = 250_000            # "early" = touched it before this
WHALE_PCT = 0.015               # 1.5% of supply in one tx
BRIDGE_WINDOW = 3600

TRANSFER = w3.keccak(text="Transfer(address,address,uint256)").hex()

# ---- FILL FROM THE VERIFIED PONS V2 CONTRACT ------------------------------
FACTORY = os.environ.get("PONS_FACTORY", "")
EV_LAUNCH = w3.keccak(text="TokenLaunched(address,address,string,uint256)").hex()
EV_BUY = w3.keccak(text="Bought(address,address,uint256,uint256)").hex()
EV_GRADUATE = w3.keccak(text="Graduated(address,address,uint256)").hex()
# Decoders live in one place (below) so an ABI change is 20 lines, not a rewrite.

KNOWN_BRIDGES = {
    # lowercase addr -> label. from docs.robinhood.com/chain + relay/across routers
}


# --------------------------------------------------------------- storage
def db():
    c = sqlite3.connect(DB, timeout=30)
    c.executescript("""
    CREATE TABLE IF NOT EXISTS tokens(
      contract TEXT PRIMARY KEY, ticker TEXT, deployer TEXT,
      launched_at REAL, early_at REAL, graduated INT DEFAULT 0,
      graduated_at REAL, peak_mcap REAL DEFAULT 0, resolved INT DEFAULT 0);

    CREATE TABLE IF NOT EXISTS deployers(
      addr TEXT PRIMARY KEY, launches INT DEFAULT 0, graduates INT DEFAULT 0);

    -- one row per wallet we have ever seen do something interesting
    CREATE TABLE IF NOT EXISTS wallets(
      addr TEXT PRIMARY KEY,
      label TEXT DEFAULT 'unknown',      -- whale | kol | deployer | fresh | unknown
      handle TEXT,                       -- ONLY if self-disclosed. never inferred
      first_seen REAL, last_seen REAL,
      entries INT DEFAULT 0,             -- tokens entered
      early_entries INT DEFAULT 0,       -- entered before EARLY_MCAP
      wins INT DEFAULT 0,                -- entered early AND it graduated
      avg_hold_s REAL DEFAULT 0,
      funder TEXT,                       -- first inbound address = funding root
      tracked INT DEFAULT 0);

    -- every position a wallet took, so behaviour is comparable across wallets
    CREATE TABLE IF NOT EXISTS positions(
      addr TEXT, contract TEXT,
      first_in REAL, last_out REAL,
      in_mcap REAL, out_mcap REAL,
      was_early INT DEFAULT 0,
      PRIMARY KEY(addr, contract));

    CREATE TABLE IF NOT EXISTS cursor(k TEXT PRIMARY KEY, v INT);
    """)
    return c


# --------------------------------------------------------------- blockscout
def bs_get(params, v2=None, retries=2):
    """v1 is the Etherscan-compatible surface; v2 is richer. Both free."""
    url = f"{BS.replace('/api','')}/api/v2{v2}" if v2 else BS
    for i in range(retries + 1):
        try:
            r = requests.get(url, params=None if v2 else params, timeout=15)
            if r.status_code == 429:
                time.sleep(2 + i * 3)
                continue
            j = r.json()
            return j.get("items") if v2 else j.get("result")
        except Exception as e:
            if i == retries:
                bus.emit("blockscout_error", err=str(e))
            time.sleep(1 + i * 2)
    return None


def addr_txs(addr, n=30, order="asc"):
    return bs_get({"module": "account", "action": "txlist", "address": addr,
                   "page": 1, "offset": n, "sort": order}) or []


def token_transfers(contract, n=100):
    return bs_get({"module": "account", "action": "tokentx",
                   "contractaddress": contract, "page": 1,
                   "offset": n, "sort": "asc"}) or []


def token_holders(contract):
    return bs_get(None, v2=f"/tokens/{contract}/holders") or []


# --------------------------------------------------------------- indexing
def dec_launch(lg):
    tok = Web3.to_checksum_address("0x" + lg["topics"][1].hex()[-40:])
    dep = Web3.to_checksum_address("0x" + lg["topics"][2].hex()[-40:])
    ticker = w3.codec.decode(["string", "uint256"], lg["data"])[0]
    return tok, dep, ticker


def dec_buy(lg):
    tok = Web3.to_checksum_address("0x" + lg["topics"][1].hex()[-40:])
    buyer = Web3.to_checksum_address("0x" + lg["topics"][2].hex()[-40:])
    _amt, price = w3.codec.decode(["uint256", "uint256"], lg["data"])
    return tok, buyer, price / 1e18


def scan_pons(c, frm, to):
    if not FACTORY:
        return
    logs = w3.eth.get_logs({"fromBlock": frm, "toBlock": to,
                            "address": Web3.to_checksum_address(FACTORY),
                            "topics": [[EV_LAUNCH, EV_BUY, EV_GRADUATE]]})
    for lg in logs:
        t = lg["topics"][0].hex()
        ts = w3.eth.get_block(lg["blockNumber"])["timestamp"]

        if t == EV_LAUNCH:
            tok, dep, ticker = dec_launch(lg)
            c.execute("INSERT OR IGNORE INTO tokens(contract,ticker,deployer,"
                      "launched_at) VALUES(?,?,?,?)", (tok, ticker, dep, ts))
            c.execute("INSERT INTO deployers(addr,launches) VALUES(?,1) "
                      "ON CONFLICT(addr) DO UPDATE SET launches=launches+1", (dep,))
            touch_wallet(c, dep, ts, label="deployer")
            bus.emit("token_launch", contract=tok, ticker=ticker, deployer=dep,
                     deployer_record=deployer_record(c, dep),
                     bridge_funded_60m=bridge_funded(dep))

        elif t == EV_BUY:
            tok, buyer, price = dec_buy(lg)
            mcap = price * (SUPPLY / 1e18)
            row = c.execute("SELECT peak_mcap, early_at, ticker FROM tokens "
                            "WHERE contract=?", (tok,)).fetchone()
            if not row:
                continue
            peak, early_at, ticker = row
            if mcap > peak:
                c.execute("UPDATE tokens SET peak_mcap=? WHERE contract=?", (mcap, tok))
            if early_at is None and mcap >= EARLY_MCAP:
                c.execute("UPDATE tokens SET early_at=? WHERE contract=?", (ts, tok))
                bus.emit("mcap_cross", contract=tok, ticker=ticker,
                         mcap=EARLY_MCAP, at=ts)
            record_entry(c, buyer, tok, ts, mcap, early=(early_at is None))

        elif t == EV_GRADUATE:
            tok = Web3.to_checksum_address("0x" + lg["topics"][1].hex()[-40:])
            c.execute("UPDATE tokens SET graduated=1, graduated_at=? "
                      "WHERE contract=?", (ts, tok))
            d = c.execute("SELECT deployer,ticker FROM tokens WHERE contract=?",
                          (tok,)).fetchone()
            if d:
                c.execute("UPDATE deployers SET graduates=graduates+1 WHERE addr=?",
                          (d[0],))
                bus.emit("token_graduate", contract=tok, ticker=d[1])
    c.commit()


def scan_transfers(c, frm, to):
    """Whale movement on tokens we still care about."""
    live = [r[0] for r in c.execute(
        "SELECT contract FROM tokens WHERE resolved=0 LIMIT 120")]
    if not live:
        return
    logs = w3.eth.get_logs({"fromBlock": frm, "toBlock": to,
                            "address": live, "topics": [TRANSFER]})
    for lg in logs:
        try:
            frm_a = Web3.to_checksum_address("0x" + lg["topics"][1].hex()[-40:])
            to_a = Web3.to_checksum_address("0x" + lg["topics"][2].hex()[-40:])
            amt = int(lg["data"].hex(), 16)
        except Exception:
            continue
        pct = amt / SUPPLY
        if pct < WHALE_PCT:
            continue
        tok = Web3.to_checksum_address(lg["address"])
        tracked = c.execute("SELECT label, handle FROM wallets WHERE addr=? "
                            "AND tracked=1", (to_a,)).fetchone()
        bus.emit("whale_transfer", contract=tok, frm=frm_a, to=to_a,
                 pct_supply=round(pct * 100, 3),
                 tx=lg["tran

tg_bridge.py

"""
tg_bridge.py — Telegram as BLOB's control plane. Nothing more.

CHIEF signs and submits its own transactions. This file never touches a key,
never places an order, never sits in the hot path. It does three things:

  1. tells you what CHIEF already did
  2. asks for APPROVE only when torah.md says to ask
  3. gives you a kill switch that works from a train, in one tap

    CHIEF ──signs & submits──► Robinhood Chain
      │
      └──card──► orders.jsonl ──► tg_bridge ──► phone
                 control.json ◄──── your taps

DEFAULT IS AUTO. A trade that passed every gate in torah.md does not need
your permission; that is what the gates were for. You get a card after the
fact with a KILL button on it.
"""

import json, time, os, threading
from pathlib import Path

import requests
import yaml

import bus

TOKEN = os.environ["TG_BOT_TOKEN"]          # revocable. never a user session.
CHAT = str(os.environ["TG_CHAT_ID"])
API = f"https://api.telegram.org/bot{TOKEN}"

TORAH = bus.ROOT / "torah.md"
DECISIONS = bus.STATE / "decisions.jsonl"

# hot-tier social noise only. warm and cold chatter stays in the logs, or you
# will train yourself to ignore your own phone inside two days.
ALERT_EVENTS = {"velocity_spike", "follow", "wallet_signal", "cluster_detected"}
HOT_ONLY = {"velocity_spike", "follow"}


# --------------------------------------------------------------- config
def cfg():
    """Execution policy lives in torah.md, not here. This file has no opinions."""
    d = {"mode": "auto", "auto_below_usd": 400, "ask_above_usd": 400,
         "approval_timeout_s": 90, "on_timeout": "skip"}
    try:
        text = TORAH.read_text()
        block = text.split("yaml")[1].split("")[0] if "yaml" in text else text
        d.update((yaml.safe_load(block) or {}).get("execution", {}))
    except Exception:
        pass
    return d


def control():
    c = bus.control()
    c.setdefault("mode", "run")
    c.setdefault("manual", cfg()["mode"] == "manual")
    return c


def set_control(**kw):
    c = control()
    c.update(kw)
    c["updated"] = time.time()
    bus.CONTROL.write_text(json.dumps(c, indent=2))
    return c


# --------------------------------------------------------------- telegram
def send(text, buttons=None):
    body = {"chat_id": CHAT, "text": text, "parse_mode": "HTML",
            "disable_web_page_preview": True}
    if buttons:
        body["reply_markup"] = json.dumps({"inline_keyboard": buttons})
    try:
        return requests.post(f"{API}/sendMessage", json=body, timeout=10).json()
    except Exception as e:
        bus.emit("tg_error", err=str(e))       # never raise into the loop
        return None


def ack(cb_id, text):
    try:
        requests.post(f"{API}/answerCallbackQuery",
                      json={"callback_query_id": cb_id, "text": text}, timeout=8)
    except Exception:
        pass


KILL = [[{"text": "KILL ALL", "callback_data": "kill"},
         {"text": "HALT NEW", "callback_data": "halt"}]]


def approve_kb(oid):
    return [[{"text": "APPROVE", "callback_data": f"ok:{oid}"},
             {"text": "SKIP", "callback_data": f"no:{oid}"}],
            [{"text": "KILL ALL", "callback_data": "kill"}]]


# --------------------------------------------------------------- cards
def fmt_gates(g):
    return "\n".join(f"  {'PASS' if v else 'FAIL'}  {k}" for k, v in g.items())


def card_entry(o, pending):
    head = ("ENTRY — NEEDS APPROVAL" if pending else "FILLED")
    tx = o.get("tx")
    return (f"<b>BLOB · {head}</b>\n"
            f"<code>{o['ticker']}</code>  {o['contract'][:12]}…\n\n"
            f"score      <b>{o['score']}</b> / {o['threshold']}\n"
            f"indep nodes {o.get('independent','—')}   "
            f"coord {o.get('coordination','—')}\n"
            f"mcap       {o.get('mcap','—')}\n"
            f"size       <b>${o['size_usd']:,.0f}</b>  "
            f"({o.get('risk_pct','—')}% of book)\n"
            f"audit      {o.get('audit','—')}\n\n"
            f"<b>gates</b>\n<pre>{fmt_gates(o.get('gates', {}))}</pre>"
            + (f"\ntx <code>{tx[:14]}…</code>" if tx else "")
            + (f"\nladder {o.get('ladder','—')}" if not pending else "")
            + ("\n\n<i>auto — already submitted</i>" if not pending else
               f"\n\n<i>expires in {cfg()['approval_timeout_s']}s → skip</i>"))


def card_exit(o):
    p = o["pnl"]
    s = "+" if p >= 0 else "−"
    return (f"<b>BLOB · {o['pct']}% SOLD</b>\n"
            f"<code>{o['ticker']}</code> @ {o['mcap']}\n"
            f"trigger: {o['trigger']}\n"
            f"realised <b>{s}${abs(p):,.2f}</b>   "
            f"session {s}${abs(o.get('session_total', p)):,.2f}")


def card_blocked(o):
    return (f"<b>BLOB · BLOCKED</b>\n<code>{o['ticker']}</code>\n"
            f"{o['reason']}\nscore {o.get('score','—')} — no entry.")


# --------------------------------------------------------------- routing
PENDING = {}          # order_id -> deadline
OPEN = {}             # ticker -> size_usd


def needs_approval(o):
    """The only two reasons to ask. Everything else fires and reports."""
    c = cfg()
    if control()["manual"]:
        return True
    return float(o.get("size_usd", 0)) > float(c["ask_above_usd"])


def decide(oid, verdict):
    with DECISIONS.open("a") as f:
        f.write(json.dumps({"id": oid, "decision": verdict,
                            "ts": time.time()}) + "\n")
    PENDING.pop(oid, None)


def push(o):
    kind = o.get("kind")

    if kind == "entry":
        if needs_approval(o):
            oid = o["id"]
            PENDING[oid] = time.time() + cfg()["approval_timeout_s"]
            send(card_entry(o, pending=True), approve_kb(oid))
        else:
            OPEN[o["ticker"]] = o.get("size_usd", 0)
            send(card_entry(o, pending=False), KILL)

    elif kind == "exit":
        if o.get("pct", 0) >= 100 or o.get("closed"):
            OPEN.pop(o["ticker"], None)
        send(card_exit(o))

    elif kind == "blocked":
        send(card_blocked(o))

    elif kind == "submit_failed":
        send(f"<b>BLOB · SUBMIT FAILED</b>\n<code>{o['ticker']}</code>\n"
             f"{o.get('err','')}\n<i>not retried — duplicate buy is worse "
             f"than a missed one</i>", KILL)


def expire_pending():
    """No answer is not a yes. Never fire on a timeout."""
    now = time.time()
    for oid, deadline in list(PENDING.items()):
        if now > deadline:
            decide(oid, cfg()["on_timeout"])
            send(f"<b>expired</b> — <code>{oid}</code> skipped, no answer in "
                 f"{cfg()['approval_timeout_s']}s")


# --------------------------------------------------------------- inbound
HELP = ("<b>BLOB control</b>\n"
        "/auto — fire without asking (default)\n"
        "/manual — approve every entry\n"
        "/kill — flatten everything, stop\n"
        "/halt — keep ladders, no new entries\n"
        "/resume — un-trip\n"
        "/status — mode, open positions")


def status():
    c = control()
    k = cfg()
    body = (f"mode <b>{c['mode']}</b>\n"
            f"entries <b>{'MANUAL' if c['manual'] else 'AUTO'}</b>"
            f"  <i>(ask above ${k['ask_above_usd']:,})</i>\n"
            f"open <b>{len(OPEN)}</b>"
            + (f" — {', '.join(OPEN)}" if OPEN else "")
            + f"\npending approval <b>{len(PENDING)}</b>")
    return body


def handle(u):
    if "callback_query" in u:
        cb = u["callback_query"]
        d, cid = cb["data"], cb["id"]

        if d == "kill":
            set_control(mode="kill"); ack(cid, "KILL SENT")
            send("<b>KILL SWITCH ENGAGED</b>\nEverything flattens on CHIEF's "
                 "next tick. No new entries until /resume.")
        elif d == "halt":
            set_control(mode="halt"); ack(cid, "halted")
            send("<b>HALTED</b>\nOpen ladders keep running. No new entries.")
        elif d.startswith(("ok:", "no:")):
            v, oid = d.split(":", 1)
            decide(oid, "approve" if v == "ok" else "skip")
            ack(cid, "approved" if v == "ok" else "skipped")
            send(("approved <code>%s</code>" % oid) if v == "ok"
                 else ("skipped <code>%s</code>" % oid))
        return

    msg = (u.get("message", {}).get("text") or "").strip().lower()
    if msg == "/auto":
        set_control(manual=False)
        send("entries <b>AUTO</b> — gates decide, you get told after.")
    elif msg == "/manual":
        set_control(manual=True)
        send("entries <b>MANUAL</b> — every entry waits for APPROVE.")
    elif msg == "/kill":
        set_control(mode="kill"); send("<b>KILL SWITCH ENGAGED</b>")
    elif msg == "/halt":
        set_control(mode="halt"); send("<b>HALTED</b>")
    elif msg == "/resume":
        set_control(mode="run"); send("resumed · mode <b>run</b>")
    elif msg == "/status":
        send(status())
    elif msg in ("/start", "/help"):
        send(HELP, KILL)


# --------------------------------------------------------------- loops
def poll_updates():
    """Own thread. A slow getUpdates must never delay an order card."""
    offset = 0
    while True:
        try:
            r = requests.get(f"{API}/getUpdates",
                             params={"offset": offset + 1, "timeout": 25},
                             timeout=35).json()
            for u in r.get("result", []):
                offset = u["update_id"]
                if str(u.get("message", {}).get("chat", {}).get("id",
                        u.get("callback_query", {}).get("message", {})
                         .get("chat", {}).get("id", ""))) != CHAT:
                    continue                      # one chat, nobody else
                try:
                    handle(u)
                except Exception as e:
                    bus.emit("tg_error", err=str(e

P.S Would be happy to answer any questions in the comments!