Building A Defi Twap Oracle Using Only Signed Event Logs

The problem I wanted to solve

I kept running into a subtle DeFi reliability issue: many “time-weighted average price” (TWAP) oracles assume they can smoothly read price state at fixed intervals. In practice, decentralized exchanges (DEXes) emit events, blocks get delayed or reordered, and the oracle logic can accidentally become sensitive to missing samples or event ordering.

So I decided to build a niche (but practical) oracle pattern: a TWAP oracle that derives its samples strictly from signed swap event logs, then verifies those logs with cryptographic signatures (no trust in my process, no reliance on local state replay).

Key idea in plain words

• A TWAP is the average of an asset’s price over time (so a single manipulation doesn’t dominate).
• Event logs are append-only records emitted by smart contracts.
• I wanted the oracle to compute TWAP from swap events as data, not as an internal “current price” variable.
• I also wanted a cryptographic trust model: each event sample is accepted only if it carries a signature from a known signer (think: an attestation key operated by an oracle operator).

This is the core of digital provenance for market data: you can prove where the samples came from and that they weren’t silently altered.

---

What I built: “SignedLogTWAP” (event-driven TWAP)

At a high level, the system has two parts:

1. An off-chain collector that receives swap events, groups them into time windows, and verifies signatures.
2. A deterministic TWAP calculator that only uses the verified samples to compute the average.

For the demo below, I’m going to simulate everything locally in Python:

• fake swap events
• fake signing keys
• signature verification
• TWAP computation

In a real deployment, the “signed event logs” would be produced by a signer contract or oracle attestation service.

---

Step 1: Define swap events and a signer signature

A swap event needs enough data to compute a price. For simplicity, I’ll use:

• timestamp (seconds)
• base_amount and quote_amount as integers
• an attestation signature over those fields

Here’s the full working Python example.

import time
import json
from dataclasses import dataclass
from typing import List, Tuple

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey


@dataclass(frozen=True)
class SwapEvent:
    timestamp: int          # unix seconds
    base_amount: int       # e.g., how many BASE tokens traded
    quote_amount: int      # e.g., how many QUOTE tokens received
    signature: bytes       # signature of the event payload


def price_from_event(e: SwapEvent) -> float:
    """
    Price (quote per base) for this swap.
    If base_amount == 0, the event is invalid (division by zero).
    """
    if e.base_amount == 0:
        raise ValueError("base_amount cannot be 0")
    return e.quote_amount / e.base_amount


def event_payload_bytes(e: SwapEvent) -> bytes:
    """
    Deterministic payload encoding.
    TWAP logic must not depend on float/string formatting quirks.
    """
    payload = {
        "timestamp": e.timestamp,
        "base_amount": e.base_amount,
        "quote_amount": e.quote_amount,
    }
    return json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")


def sign_event(priv: Ed25519PrivateKey, e_no_sig: SwapEvent) -> SwapEvent:
    payload = event_payload_bytes(e_no_sig)
    sig = priv.sign(payload)
    return SwapEvent(
        timestamp=e_no_sig.timestamp,
        base_amount=e_no_sig.base_amount,
        quote_amount=e_no_sig.quote_amount,
        signature=sig
    )


def verify_event(pub: Ed25519PublicKey, e: SwapEvent) -> bool:
    try:
        pub.verify(e.signature, event_payload_bytes(e))
        return True
    except Exception:
        return False


def twap_from_events(events: List[SwapEvent], window_start: int, window_end: int, min_samples: int = 2) -> float:
    """
    Computes a TWAP over [window_start, window_end] using verified events only.

    This demo uses a simple approach:
    - Keep events strictly inside the interval.
    - Sort by timestamp.
    - Weight each event by the time until the next event.
      (Last sample gets weighted by the remaining time until window_end.)
    """
    filtered = [e for e in events if window_start <= e.timestamp <= window_end]
    if len(filtered) < min_samples:
        raise ValueError(f"Not enough samples: {len(filtered)}")

    filtered.sort(key=lambda x: x.timestamp)

    total_weighted = 0.0
    total_weight = 0

    for i, e in enumerate(filtered):
        t0 = e.timestamp
        if i + 1 < len(filtered):
            t1 = filtered[i + 1].timestamp
        else:
            t1 = window_end

        if t1 <= t0:
            # Ignore non-monotonic timestamps; real systems should handle this earlier.
            continue

        dt = t1 - t0
        p = price_from_event(e)
        total_weighted += p * dt
        total_weight += dt

    if total_weight == 0:
        raise ValueError("Total weight is zero; timestamps may be invalid")

    return total_weighted / total_weight

What each block is doing (and why)

• event_payload_bytes() makes the signing/verifying payload deterministic via sorted JSON and no whitespace.
• sign_event() signs only the payload fields that matter for price.
• verify_event() ensures the event is authentic before it can influence TWAP.
• twap_from_events():
  • filters to the requested time window
  • sorts by timestamp
  • weights each sample by the time until the next sample

This weighting step is what makes it “time-weighted” rather than just averaging prices.

---

Step 2: Run the demo with realistic “event gaps”

Now I simulate:

• a signer key
• a handful of swap events at uneven times (some gaps)
• one tampered event (wrong signature)
• compute TWAP from only verified samples

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

# --- generate signer keys ---
priv = Ed25519PrivateKey.generate()
pub = priv.public_key()

now = int(time.time())
# Build events at irregular intervals (simulate real-world block/event timing)
base_events = [
    # valid event
    (now - 120, 1000, 2500),
    # valid event
    (now - 90,  800,  2020),
    # tampering attempt: will be signed incorrectly
    (now - 60,  900,  2600),
    # valid event
    (now - 30,  700,  1890),
    # valid event (late sample)
    (now - 10,  500,  1350),
]

# Create signed events
signed_events: List[SwapEvent] = []
for ts, base_amt, quote_amt in base_events:
    e_no_sig = SwapEvent(timestamp=ts, base_amount=base_amt, quote_amount=quote_amt, signature=b"")
    signed_events.append(sign_event(priv, e_no_sig))

# Tamper the third event's payload without updating the signature
tampered = signed_events[2]
tampered_t = SwapEvent(
    timestamp=tampered.timestamp,
    base_amount=tampered.base_amount,
    quote_amount=tampered.quote_amount + 9999,  # malicious change
    signature=tampered.signature,                  # keep old signature
)
signed_events[2] = tampered_t

# Verify and keep only valid events
verified = [e for e in signed_events if verify_event(pub, e)]

print("All events:", len(signed_events))
print("Verified events:", len(verified))
for e in sorted(verified, key=lambda x: x.timestamp):
    print(" - t=", e.timestamp, "price=", price_from_event(e))

# Compute TWAP in a window
window_start = now - 100
window_end = now - 5

twap = twap_from_events(verified, window_start, window_end)
print("TWAP:", twap)

What this proves

• Even if an attacker injects a modified event, the signature check prevents it from affecting the TWAP.
• Irregular timing is handled via time weighting (not just “average all prices”).

In my weekend experiments, this pattern was the difference between “it mostly works” and “it’s robust when event arrival isn’t polite.”

---

Step 3: Make the weighting deterministic and on-chain friendly (no floats)

In a real DeFi system, you don’t want float math. The typical on-chain approach is to compute using fixed-point integers.

Below is a drop-in integer TWAP calculator:

• prices as rational numbers: quote_amount / base_amount
• exact numerator/denominator arithmetic using integer weighting

We can keep it simple by scaling to a fixed-point format (e.g., 1e18).

from fractions import Fraction

SCALE = 10**18

def twap_from_events_fixed(
    events: List[SwapEvent],
    window_start: int,
    window_end: int,
    min_samples: int = 2,
    scale: int = SCALE
) -> int:
    filtered = [e for e in events if window_start <= e.timestamp <= window_end]
    if len(filtered) < min_samples:
        raise ValueError(f"Not enough samples: {len(filtered)}")

    filtered.sort(key=lambda x: x.timestamp)

    total_weight = 0
    total_weighted_price_num = 0  # scaled numerator

    for i, e in enumerate(filtered):
        t0 = e.timestamp
        t1 = filtered[i + 1].timestamp if i + 1 < len(filtered) else window_end

        if t1 <= t0:
            continue

        dt = t1 - t0

        # price = quote_amount / base_amount
        # scaled_price = floor(price * scale)
        # weighted contribution = scaled_price * dt
        if e.base_amount == 0:
            raise ValueError("base_amount cannot be 0")

        scaled_price = (e.quote_amount * scale) // e.base_amount

        total_weighted_price_num += scaled_price * dt
        total_weight += dt

    if total_weight == 0:
        raise ValueError("Total weight is zero; timestamps may be invalid")

    # TWAP scaled integer
    return total_weighted_price_num // total_weight


# Example usage (reusing verified from earlier):
twap_fixed = twap_from_events_fixed(verified, window_start, window_end)
print("TWAP fixed-point:", twap_fixed)
print("TWAP approx:", twap_fixed / SCALE)

Why the fixed-point version matters

• It makes the oracle logic consistent with smart contract arithmetic constraints.
• It eliminates rounding surprises from float representation differences between platforms.

---

Where this fits in DeFi (niche but useful)

This “signed event log TWAP” pattern is especially relevant when:

• you need strong data provenance for pricing inputs
• you want to tolerate missing or delayed events without silently using unverified state
• you want an oracle that can be audited as: “Given these signed samples, TWAP equals X.”

That audit trail is a big deal in DeFi, where small oracle weaknesses can cascade into large liquidation or manipulation events.

---

Conclusion

I built a compact “SignedLogTWAP” prototype that computes TWAP strictly from cryptographically verified swap event logs, then uses time-weighted sampling based on event timestamps (with an integer fixed-point variant for on-chain realism). The main takeaway from tinkering with this was that robust DeFi price oracles are less about fancy math and more about trust boundaries: if every price sample is authenticated and deterministically weighted, the resulting TWAP becomes both reliable and provable.