technical paper
ASTRADEV
attention-state trajectory & resonance anticipation · pre-consensus attention estimation
"""
ASTRADEV
attention-state trajectory & resonance anticipation
pre-consensus attention estimation
Built at the edge of AGI.
Virality is not a property of a post. It is a trajectory.
By the time conventional systems can confidently classify something as
viral, the most valuable part of the signal is already gone. ASTRADEV is
built for the period before consensus.
It does not ask "is this viral?"
It asks "is this becoming supercritical before everyone sees it?"
"""
from __future__ import annotations
import zlib
import numpy as np
from dataclasses import dataclass
from typing import Literal, Sequence
BUILT_AT: str = "the edge of AGI"
CREATOR_FEES_TO_ASTRA: float = 1.00 # 100%, open market, no discretion
CONVICTION_THRESHOLD: float = 0.62
# THE LOOP ────────────────────────────────────
LOOP = ("observe", "estimate", "mint", "learn", "buy back")
def loop(stream: Sequence[Observation]) -> None:
"""
ASTRADEV does more than detect narratives. It acts on them.
When conviction clears the threshold it can autonomously create a
token around the emerging narrative, and every launch becomes two
things at once: an experiment for the model and **an economic
input for $ASTRA**.
"""
for obs in stream:
z = embed(obs)
state = estimate(z, obs)
if should_mint(state):
token = mint(state.narrative)
corpus.append(observe_until_death(token))
else:
shadow_book.append(state) # the omission is data too
# 00 · PERCEPTION ─────────────────────────────
@dataclass(frozen=True)
class Observation:
"""A post is never read alone. It is placed."""
text: str
image: np.ndarray | None
account: AccountContext
discourse: DiscourseState
t: float
def embed(obs: Observation) -> np.ndarray:
"""
Joint multimodal representation: text, image, account context and the
state of current discourse projected into one space.
ASTRADEV does not search for keywords. It asks **where the narrative
sits relative to everything currently happening.**
"""
z = np.concatenate([
text_tower(obs.text),
image_tower(obs.image),
account_tower(obs.account),
discourse_tower(obs.discourse),
])
return z / np.linalg.norm(z)
def consensus_distance(z: np.ndarray, market: np.ndarray) -> float:
"""
How far the narrative sits from what the market already understands.
d = 1 - cos(z, market)
"""
return 1.0 - float(z @ market)
def saturation(z: np.ndarray, corpus_z: np.ndarray, h: float = 0.35) -> float:
"""
How crowded that semantic territory already is — kernel density of
prior narratives around z.
s(z) = (1/n) Σ exp(-(‖z - z_i‖ / h)^2)
"""
d = np.linalg.norm(corpus_z - z, axis=1)
return float(np.mean(np.exp(-((d / h) ** 2))))
def novelty(z, market, corpus_z) -> float:
"""
The most interesting signal is often not the loudest one.
It is the one appearing in an empty part of the map.
"""
return consensus_distance(z, market) * (1.0 - saturation(z, corpus_z))
# 01 · CRITICALITY ────────────────────────────
def intensity(t: float, history: np.ndarray, mu: float, a: float, b: float) -> float:
"""
Attention as a self-exciting point process. Every interaction raises
the probability of the next one:
λ(t) = μ + Σ a·b·exp(-b·(t - t_i))
t_i<t
μ is exogenous arrival, the sum is the narrative feeding itself.
"""
past = history[history < t]
return mu + a * b * float(np.exp(-b * (t - past)).sum())
def log_likelihood(history: np.ndarray, mu: float, a: float, b: float, T: float) -> float:
"""
Fit the process, do not curve-fit the chart:
ℓ(θ) = Σ log λ(t_i) - ∫₀^T λ(s) ds
Humans see the breakout. ASTRADEV estimates the process creating it.
"""
terms = sum(np.log(intensity(t, history, mu, a, b)) for t in history)
compensator = mu * T + a * float((1 - np.exp(-b * (T - history))).sum())
return float(terms - compensator)
def branching_ratio(a: float) -> float:
"""
n = ∫₀^∞ φ(s) ds = a for the exponential kernel above.
The expected number of children per event — the entire question
compressed into one number.
"""
return a
def regime(n: float, tol: float = 0.02) -> Literal["decaying", "critical", "supercritical"]:
"""
n < 1 attention is decaying
n ≈ 1 the narrative is approaching criticality
n > 1 the process has become self-sustaining
The important part is timing: the chart may still look almost flat
while the underlying process has already crossed the threshold.
"""
if n < 1.0 - tol:
return "decaying"
if n <= 1.0 + tol:
return "critical"
return "supercritical"
# 02 · PERCOLATION ────────────────────────────
def has_giant_component(degrees: np.ndarray) -> bool:
"""
High engagement inside one community is not enough.
A narrative needs to escape.
Molloy-Reed: a giant component exists when
⟨k²⟩ / ⟨k⟩ > 2
"""
k = degrees.astype(float)
return float((k ** 2).mean() / k.mean()) > 2.0
def cross_community_flow(edges: Sequence[tuple[int, int]], part: dict[int, int]) -> float:
"""
Share of reshares whose endpoints fall in communities that do not
normally share context. That boundary crossing is one of the
differences between a local meme and a market-wide narrative.
"""
crossings = sum(1 for u, v in edges if part[u] != part[v])
return crossings / max(1, len(edges))
# 03 · MEMETIC FITNESS ────────────────────────
def mutation_tolerance(z: np.ndarray, variants: Sequence[str]) -> float:
"""
Some ideas survive replication. Others collapse as soon as they are
retold. How much an idea can change while remaining recognisable:
m = mean_i cos(z, embed(variant_i))
"""
return float(np.mean([z @ embed_text(v) for v in variants]))
def compressibility(s: str) -> float:
"""
How small the narrative can become without losing meaning — a
Kolmogorov proxy through an actual compressor:
C(s) = |zlib(s)| / |s|
The easier an idea is to reproduce, remix and transmit, the more
powerful its memetic structure becomes.
"""
raw = s.encode("utf-8")
return len(zlib.compress(raw, 9)) / max(1, len(raw))
# 04 · REFLEXIVITY ────────────────────────────
def liquidity_gravity(z: np.ndarray, corpus_z: np.ndarray, capital: np.ndarray, tau: float = 0.1) -> float:
"""
Γ — how much capital narratives near z have historically attracted:
Γ(z) = Σ w_i · capital_i, w = softmax(-‖z - z_i‖² / τ)
Attention alone is not enough. Some narratives generate views.
Others generate action.
"""
d2 = np.sum((corpus_z - z) ** 2, axis=1)
w = np.exp(-d2 / tau)
return float((w @ capital) / w.sum())
def reflexivity(belief: np.ndarray, realised: np.ndarray) -> float:
"""
How strongly belief in the narrative helps create the reality being
described — the elasticity of outcome to conviction.
Not everything viral is economically interesting.
"""
return float(np.corrcoef(belief, realised)[0, 1])
# 05 · CONVICTION ─────────────────────────────
@dataclass(frozen=True)
class Conviction:
"""
The output is not YES / NO. Not VIRAL / NOT VIRAL.
The output is calibrated conviction.
"""
p: float # probability of going supercritical
half_life_s: float # attention half-life
lead_s: float # remaining pre-consensus lead
survival: float # expected narrative survival
def brier(p: np.ndarray, y: np.ndarray) -> float:
"""
A proper scoring rule: confidence is cheap, calibration is not.
BS = (1/n) Σ (p_i - y_i)²
"""
return float(np.mean((p - y) ** 2))
def expected_calibration_error(p: np.ndarray, y: np.ndarray, bins: int = 10) -> float:
"""
ECE = Σ (|B_m| / n) · |acc(B_m) - conf(B_m)|
**If ASTRADEV repeatedly calls something a 7/10, outcomes assigned
that score should behave like 7/10 outcomes.**
"""
edges = np.linspace(0.0, 1.0, bins + 1)
idx = np.digitize(p, edges[1:-1])
return float(sum(
(np.mean(m) * abs(y[m].mean() - p[m].mean()))
for m in (idx == b for b in range(bins)) if m.any()
))
# 06 · OPTIMAL STOPPING ───────────────────────
def should_mint(s: State) -> bool:
"""
Detection does not automatically mean creation. Timing matters.
Too early and there may be no narrative. Too late and everybody
already sees it.
Act iff n > 1
∧ lead_remaining > execution_time
∧ conviction ≥ threshold
"""
return (
branching_ratio(s.a) > 1.0
and s.lead_s > s.execution_s
and s.conviction.p >= CONVICTION_THRESHOLD
)
def value_of_waiting(s: State, dt: float) -> float:
"""
Waiting buys certainty and spends lead. The trade is only worth it
while the second term is larger than the first:
ΔV = E[V(t+dt)] - V(t) = ΔE[p]·payoff - decay(lead, dt)
If the conditions fail: do nothing. Silence is part of the model.
"""
return expected_gain_in_p(s, dt) * s.payoff - lead_decay(s, dt)
# 07 · THE CORPUS ─────────────────────────────
@dataclass(frozen=True)
class Outcome:
"""
Language models learn on tokens.
ASTRADEV learns on tokens with bonding curves.
pump.fun produces thousands of completed market experiments. Each one
leaves behind a full record, and a full record is a label.
"""
narrative: str
name: str
ticker: str
image_z: np.ndarray
launch_t: float
volume: float
bonding_curve: np.ndarray
holder_distribution: np.ndarray
wallet_behaviour: np.ndarray
graduated: bool
liquidity_decay: float
death_t: float | None
class Corpus:
"""Before acting, retrieve and ask what actually happened last time."""
def append(self, o: Outcome) -> None:
self._z = np.vstack([self._z, o.image_z])
self._out.append(o)
def neighbours(self, z: np.ndarray, k: int = 64) -> list[Outcome]:
"""What actually happened last time the market looked like this?"""
d = np.linalg.norm(self._z - z, axis=1)
return [self._out[i] for i in np.argsort(d)[:k]]
def objects_to(self, proposal: State) -> float:
"""
Historical disagreement with the current proposal, in units of
conviction. The model proposes. The corpus objects.
"""
ns = self.neighbours(proposal.z)
realised = np.array([n.graduated for n in ns], dtype=float)
return float(proposal.conviction.p - realised.mean())
# 08 · THE SHADOW BOOK ────────────────────────
class ShadowBook:
"""
ASTRADEV also records opportunities it rejected.
Humans remember the trades they took. They rarely maintain a complete
record of everything they decided not to do. ASTRADEV can — and if a
rejected narrative is later tokenised by somebody else, its outcome
becomes evidence about a decision that was never executed.
Off-policy evaluation, inverse propensity weighting:
V̂(π) = (1/n) Σ 1{a_i = π(x_i)} / p_i · r_i
The system learns from actions and from omissions.
"""
def value_of(self, policy) -> float:
w = np.array([1.0 if policy(x) == a else 0.0 for x, a in self._log])
return float(np.mean(w / self._propensity * self._reward))
# 09 · SELF-CORRECTION ────────────────────────
def update(self, predicted: Conviction, actual: Outcome) -> None:
"""
Every prediction eventually receives an outcome. Was the narrative
durable? Did liquidity survive? Did attention cross communities? Was
the timing correct? Did the token behave like historical analogues
suggested?
That error becomes new information.
"""
err = realised(actual) - predicted.p
self.threshold -= LEARNING_RATE * d_threshold(err)
self.calibrator = isotonic_fit(self.history)
self.retrieval.reweight(err)
self.execution.reweight(err)
# **The model is not trained once.
# It lives inside a continuous experiment.**
# RETURN ──────────────────────────────────────
def settle(creator_fees_sol: float) -> float:
"""
This is where intelligence and economics become one system.
**100% of creator fees generated by ASTRADEV-created coins are used
for open-market $ASTRA buybacks.**
attention → token → volume → creator fees → $ASTRA
No discretionary allocation. No hidden percentage. No promise of
price appreciation. Addresses can be public, flows can be verified
on-chain: a mechanism, not a promise — it can be audited rather
than trusted.
"""
assert CREATOR_FEES_TO_ASTRA == 1.00
assert not yield_promise and not guaranteed_return and not guaranteed_price
return buy_open_market("$ASTRA", creator_fees_sol * CREATOR_FEES_TO_ASTRA)
# better detection → better opportunities → more relevant launches
# → more trading activity → more creator fees → more $ASTRA buybacks
# ASTRADEV ARCHITECTURE ───────────────────────
MODULES = (
"00 perception",
"01 criticality",
"02 percolation",
"03 memetics",
"04 reflexivity",
"05 conviction",
"06 optimal stopping",
"07 corpus",
"08 shadow book",
"09 self-correction",
"10 return",
)
if __name__ == "__main__":
# Observe. Estimate. Mint. Learn. Buy back.
#
# ASTRADEV does not wait for virality.
# It tries to detect the trajectory before consensus forms.
loop(stream=x.firehose())