레시피: 동결 기록 원장을 폴링하고 새 기록을 알리기

상태 하나 — 처리한 가장 최근 기록의 (block_time, tx_hash, log_index) — 만 가진 폴러입니다. 30초마다(이 경로의 max-age) /v1/freezes/acts를 읽고, 이미 본 지점에 도달할 때까지 next_cursor를 따라 순회하며, subject가 관심 목록에 있는 모든 새 기록, 또는 임계값을 넘는 모든 소각에 대해 알림을 보냅니다. 프로그램은 새 기록마다 한 줄을 출력하고 상태를 파일에 보관하므로, 재시작해도 다시 알리지 않습니다. 텔레그램 봇은 사람을 위해 같은 일을 합니다(/v1/subscriptions가 위치를 알려 줍니다). 이것은 직접 작성하는 코드용 버전입니다.

루프

1. 필터(chain=, token= — 또는 모든 네트워크라면 필터 없음)로 원장의 첫 페이지를 읽습니다.
2. 가장 최근 행부터 앞으로 순회합니다. 이미 처리한 첫 행(block_time ≤ 저장된 표시이고 같은 tx_hash/log_index) 또는 페이지 끝에서 멈춥니다. 페이지 전체가 새 행인 경우에만 next_cursor를 따라갑니다.
3. 모든 새 기록에 대해: subject가 관심 대상이거나, action이 seize이고 amount_usd6가 임계값을 넘으면 알림을 보냅니다.
4. 가장 최근 행의 표시를 저장합니다. 30초 대기합니다.

원장은 확정 전의 기록을 finality: "pending"으로 나열합니다. 이 레시피는 그 기록에 대해서도 알림을 보내고 줄에 확정성을 표시합니다. 이미 실행된 동결 제안은 블록이 확정되기 전에 알아 둘 가치가 있기 때문입니다.

curl

#!/bin/sh
# freezes.sh — every 30 s, print the acts newer than the last one seen (state in ./last_act)
B=https://stablescan.achivx.com; STATE=./last_act; WATCH="${WATCH:-0xb84e20794fd31fcb93e4b621440fbc5af5ef8f60}"
while :; do
  last=$(cat $STATE 2>/dev/null)
  page=$(curl -sf "$B/v1/freezes/acts?limit=100") || { sleep 30; continue; }
  echo "$page" | jq -r --arg last "$last" --arg watch "$WATCH" '
    [.items[] | select((.block_time + "/" + .tx_hash + "/" + (.log_index|tostring)) > $last)] | reverse[]
    | "\(.block_time) \(.chain_slug) \(.token) \(.action) \(.subject) \(if .amount_usd6 then "$" + (.amount_usd6/1e6|round|tostring) else "" end) \(.finality)"
      + (if (.subject | ascii_downcase) == ($watch | ascii_downcase) then "  <-- WATCHED" else "" end)'
  echo "$page" | jq -r '.items[0] | .block_time + "/" + .tx_hash + "/" + (.log_index|tostring)' > $STATE
  sleep 30
done

셸 버전은 한 페이지만 비교합니다. 아래 세 가지는 표시에 도달할 때까지 앞으로 페이지를 넘깁니다.

Python

import json, os, sys, time, urllib.error, urllib.parse, urllib.request

BASE = "https://stablescan.achivx.com"
STATE = "last_act.json"
WATCH = {a.lower() for a in os.environ.get("WATCH", "").split(",") if a}
SEIZE_ALERT_USD = float(os.environ.get("SEIZE_ALERT_USD", "100000"))

def scan(path, **params):
    url = BASE + path + ("?" + urllib.parse.urlencode(params) if params else "")
    with urllib.request.urlopen(urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "freezes/1.0"}), timeout=30) as r:
        return json.load(r)

def mark(act): return (act["block_time"], act["tx_hash"], act["log_index"])

def new_acts(last, **filters):
    '''every act newer than `last`, oldest first; pages forward until it meets the mark'''
    out, cursor = [], None
    while True:
        page = scan("/v1/freezes/acts", limit=200, **filters, **({"cursor": cursor} if cursor else {}))
        for act in page["items"]:
            if last and mark(act) <= last:
                return list(reversed(out))
            out.append(act)
        cursor = page.get("next_cursor")
        if not cursor or not last:               # first run: one page is enough to set the mark
            return list(reversed(out))

def alert(act):
    usd = act.get("amount_usd6", 0) / 1e6
    watched = act["subject"].lower() in WATCH
    big = act["action"] == "seize" and usd >= SEIZE_ALERT_USD
    line = f'{act["block_time"]} {act["chain_slug"]} {act["token"]} {act["action"]} {act["subject"]} {f"${usd:,.0f}" if usd else ""} {act["finality"]}'
    if watched or big:
        print("ALERT", line, "(watched)" if watched else "", flush=True)   # send it wherever alerts go
    else:
        print(line, flush=True)

def main(**filters):
    last = tuple(json.load(open(STATE))) if os.path.exists(STATE) else None
    while True:
        try:
            acts = new_acts(last, **filters)
            for act in acts:
                alert(act)
            if acts:
                last = mark(acts[-1]); json.dump(last, open(STATE, "w"))
        except urllib.error.HTTPError as e:
            print("scanner", e.code, e.headers.get("Retry-After"), file=sys.stderr)
            time.sleep(float(e.headers.get("Retry-After") or 30))
        time.sleep(30)                                 # the ledger's max-age

if __name__ == "__main__":
    main(**dict(a.split("=", 1) for a in sys.argv[1:]))   # e.g. chain=tron token=USDT

Go

package main

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"strings"
	"time"
)

const base = "https://stablescan.achivx.com"

type act struct {
	ChainSlug  string  `json:"chain_slug"`
	Token      string  `json:"token"`
	Subject    string  `json:"subject"`
	Action     string  `json:"action"`
	AmountUSD6 float64 `json:"amount_usd6"`
	TxHash     string  `json:"tx_hash"`
	LogIndex   int     `json:"log_index"`
	BlockTime  string  `json:"block_time"`
	Finality   string  `json:"finality"`
}

func (a act) mark() string { return fmt.Sprintf("%s/%s/%06d", a.BlockTime, a.TxHash, a.LogIndex) }

func page(q url.Values) ([]act, string, error) {
	res, err := http.Get(base + "/v1/freezes/acts?" + q.Encode())
	if err != nil {
		return nil, "", err
	}
	defer res.Body.Close()
	if res.StatusCode != 200 {
		return nil, "", fmt.Errorf("HTTP %d, Retry-After %s", res.StatusCode, res.Header.Get("Retry-After"))
	}
	var p struct {
		Items      []act  `json:"items"`
		NextCursor string `json:"next_cursor"`
	}
	err = json.NewDecoder(res.Body).Decode(&p)
	return p.Items, p.NextCursor, err
}

// newActs returns every act newer than last, oldest first.
func newActs(last string, filters url.Values) ([]act, error) {
	var out []act
	q := url.Values{"limit": {"200"}}
	for k, v := range filters {
		q[k] = v
	}
	for {
		items, next, err := page(q)
		if err != nil {
			return nil, err
		}
		for _, a := range items {
			if last != "" && a.mark() <= last {
				return reverse(out), nil
			}
			out = append(out, a)
		}
		if next == "" || last == "" {
			return reverse(out), nil
		}
		q.Set("cursor", next)
	}
}

func reverse(xs []act) []act {
	for i, j := 0, len(xs)-1; i < j; i, j = i+1, j-1 {
		xs[i], xs[j] = xs[j], xs[i]
	}
	return xs
}

func main() {
	watch := map[string]bool{}
	for _, a := range strings.Split(os.Getenv("WATCH"), ",") {
		if a != "" {
			watch[strings.ToLower(a)] = true
		}
	}
	filters := url.Values{}
	for _, kv := range os.Args[1:] { // chain=tron token=USDT
		if k, v, ok := strings.Cut(kv, "="); ok {
			filters.Set(k, v)
		}
	}
	last := ""
	if b, err := os.ReadFile("last_act"); err == nil {
		last = strings.TrimSpace(string(b))
	}
	for {
		acts, err := newActs(last, filters)
		if err != nil {
			fmt.Fprintln(os.Stderr, "scanner:", err)
			time.Sleep(30 * time.Second)
			continue
		}
		for _, a := range acts {
			usd := a.AmountUSD6 / 1e6
			line := fmt.Sprintf("%s %s %s %s %s $%.0f %s", a.BlockTime, a.ChainSlug, a.Token, a.Action, a.Subject, usd, a.Finality)
			if watch[strings.ToLower(a.Subject)] || (a.Action == "seize" && usd >= 100000) {
				fmt.Println("ALERT", line)
			} else {
				fmt.Println(line)
			}
		}
		if len(acts) > 0 {
			last = acts[len(acts)-1].mark()
			_ = os.WriteFile("last_act", []byte(last), 0o644)
		}
		time.Sleep(30 * time.Second)
	}
}

TypeScript

import { readFile, writeFile } from "node:fs/promises";

const BASE = "https://stablescan.achivx.com";
const WATCH = new Set((process.env.WATCH ?? "").split(",").filter(Boolean).map((a) => a.toLowerCase()));
const SEIZE_ALERT_USD = Number(process.env.SEIZE_ALERT_USD ?? 100_000);
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

type Act = { chain_slug: string; token: string; subject: string; action: "freeze" | "unfreeze" | "seize"; amount_usd6?: number; tx_hash: string; log_index: number; block_time: string; finality: "final" | "pending" };
type Page = { items: Act[]; next_cursor?: string };
const mark = (a: Act) => `${a.block_time}/${a.tx_hash}/${String(a.log_index).padStart(6, "0")}`;

async function scan<T>(path: string, params: Record<string, string>): Promise<T> {
  const res = await fetch(`${BASE}${path}?${new URLSearchParams(params)}`, { headers: { accept: "application/json" } });
  if (!res.ok) throw new Error(`HTTP ${res.status}, Retry-After ${res.headers.get("retry-after")}`);
  return (await res.json()) as T;
}

async function newActs(last: string, filters: Record<string, string>): Promise<Act[]> {
  const out: Act[] = [];
  let cursor: string | undefined;
  for (;;) {
    const page = await scan<Page>("/v1/freezes/acts", { limit: "200", ...filters, ...(cursor ? { cursor } : {}) });
    for (const a of page.items) {
      if (last && mark(a) <= last) return out.reverse();
      out.push(a);
    }
    cursor = page.next_cursor;
    if (!cursor || !last) return out.reverse();
  }
}

const filters = Object.fromEntries(process.argv.slice(2).map((kv) => kv.split("=", 2) as [string, string]));   // chain=tron token=USDT
let last = await readFile("last_act", "utf8").catch(() => "");
for (;;) {
  try {
    const acts = await newActs(last.trim(), filters);
    for (const a of acts) {
      const usd = (a.amount_usd6 ?? 0) / 1e6;
      const line = `${a.block_time} ${a.chain_slug} ${a.token} ${a.action} ${a.subject} ${usd ? `$${usd.toLocaleString()}` : ""} ${a.finality}`;
      const hit = WATCH.has(a.subject.toLowerCase()) || (a.action === "seize" && usd >= SEIZE_ALERT_USD);
      console.log(hit ? `ALERT ${line}` : line);
    }
    if (acts.length) {
      last = mark(acts[acts.length - 1]);
      await writeFile("last_act", last);
    }
  } catch (e) {
    console.error("scanner:", String(e));
  }
  await sleep(30_000);
}

2026-10-24의 출력

10:50 UTC에 chain=tron token=USDT로 시작하자, 첫 페이지는 전날 저녁 21:50의 TYse3PTC…b9z4 동결에 표시를 설정했습니다. 이후 몇 분 동안의 다음 새 행들은 트론(Tron)에서의 테더(Tether) 오전 일괄 처리 — finality: "final"인 동결 — 였으며, 전날 16:51의 480,326달러와 405,413달러 소각 두 건은 이미 표시보다 이전이었습니다. WATCH=0xb84e20794fd31fcb93e4b621440fbc5af5ef8f60이고 필터가 없을 때, 그 주소에 대한 원장 행(2026-10-23 16:32:47 동결, 제안 5786)이 첫 실행에서 ALERT를 출력하는 행입니다.

운영용 폴러가 추가로 하는 두 가지

• 같은 주기로 /v1/freezes/proposals?chain=1&status=pending을 읽습니다. 테더의 기록은 이더리움(Ethereum)에 반영되기 중앙값 44분 전(트론은 25분)에 그곳에서 가려진 형태로 보입니다. 대상은 실행 시에만 공개되므로, 그곳의 알림은 누구에게가 아니라 동결이 다가온다는 것입니다.
• 알림 후에는 그 chain으로 대상의 개요를 읽습니다. 원장 행은 이벤트이고, freeze[].status는 상태입니다. 한 시간 뒤의 동결 해제는 두 번째 행이자 clear입니다.

Stablecoin scanner ACHIVX에서

• 폴러가 읽는 원장: /v1/freezes/acts?limit=100. 그 옆의 제안: /v1/freezes/proposals?chain=1&status=pending.
• 사람을 위한 알림: /v1/subscriptions에 텔레그램 봇과 시작 페이로드(동결용 f_<address>)가 나와 있습니다.

출처

• REST 인터페이스의 OpenAPI 3.1 명세(API 1.4.0)
• 에이전트 매뉴얼 — agents.md(버전 2026-10-05)
• GET /v1/freezes/acts?chain=tron&token=USDT&limit=3(2026-10-24 조회)

더 알아보기

• 관심 목록과 알림
• 스크립트는 발행사 기록을 어떻게 통보받습니까 — 관심 목록과 구독?
• 동결과 알림

관련 질문

• 발행사의 동결, 동결 해제, 소각은 어떻게 조회합니까?
• 동결이 실행되기 전에 미리 보려면 어떻게 합니까: 발행사 멀티시그의 제안
• 코드에서 피드를 페이지 단위로 읽고 응답을 캐시하려면 어떻게 합니까?

레시피 · acts → overview · 2026-10-24 작성, openapi.json, agents.md 및 당일 실제 호출 기반. 구조는 인용한 것이며, 수치는 날짜 기준이고 인덱스에 따라 바뀝니다.