AI × Crypto Synergy in Brokerage Rails: Architecting Verifiable Social Signals

// signals/ingest.ts — TypeScript 5.5+, no runtime deps required for the type layer
export type Result<T, E> = { ok: true; value: T } | { ok: false; error: E };

export type IngestError =
  | { kind: 'schema'; detail: string }
  | { kind: 'duplicate'; signalId: string }
  | { kind: 'upstream'; status: number; retryAfterMs: number };

export type ClaimType = 'ownership_stake' | 'product_launch' | 'partnership' | 'unclassified';

export interface NormalizedSignal {
  readonly signalId: string;      // sha256(`${source}|${nativeId}`)
  readonly source: 'social' | 'onchain' | 'filing';
  readonly assets: readonly string[];
  readonly claimType: ClaimType;
  readonly confidenceBps: number; // 0..10000, integer only
  readonly observedAt: string;    // RFC 3339, UTC
}

const TICKER_RE = /^[A-Z]{2,10}(?:\/USD)?$/;

export function normalizeMention(raw: unknown, now: Date = new Date()): Result<NormalizedSignal, IngestError> {
  if (typeof raw !== 'object' || raw === null) {
    return { ok: false, error: { kind: 'schema', detail: 'payload is not an object' } };
  }
  const r = raw as Record<string, unknown>;
  if (typeof r.id !== 'string' || r.id.length === 0 || r.id.length > 128) {
    return { ok: false, error: { kind: 'schema', detail: 'invalid id' } };
  }
  const text = typeof r.text === 'string' ? r.text : '';
  const observedAt = typeof r.createdAt === 'string' ? new Date(r.createdAt) : now;
  if (Number.isNaN(observedAt.getTime())) {
    return { ok: false, error: { kind: 'schema', detail: 'unparseable createdAt' } };
  }

  const tickers = Array.from(text.matchAll(/\$([A-Za-z]{2,10})/g), (m) => m[1].toUpperCase())
    .filter((t) => TICKER_RE.test(t));
  if (tickers.length === 0) {
    return { ok: false, error: { kind: 'schema', detail: 'no resolvable ticker' } };
  }

  const claimType: ClaimType = /\bstake\b|\bequity\b|\bownership\b/i.test(text)
    ? 'ownership_stake'
    : 'unclassified';

  return {
    ok: true,
    value: {
      signalId: await sha256Hex(`${String(r.source ?? 'social')}|${r.id}`),
      source: 'social',
      assets: [...new Set(tickers)],
      claimType,
      confidenceBps: 0, // assigned downstream by the classifier, never by the ingestor
      observedAt: observedAt.toISOString(),
    },
  };
}

// signals/publish.ts
export class SignalIngestor {
  constructor(
    private readonly redis: { set(...a: unknown[]): Promise<unknown>; del(k: string): Promise<unknown> },
    private readonly bus: { send(m: { topic: string; messages: unknown[] }): Promise<void> },
  ) {}

  async publish(raw: unknown): Promise<Result<NormalizedSignal, IngestError>> {
    const parsed = normalizeMention(raw);
    if (!parsed.ok) return parsed;
    const sig = parsed.value;

    const key = `signal:seen:${sig.signalId}`;
    const claimed = await this.redis.set(key, '1', 'NX', 'PX', 86_400_000); // 24h dedupe window
    if (claimed !== 'OK') return { ok: false, error: { kind: 'duplicate', signalId: sig.signalId } };

    try {
      await this.bus.send({
        topic: 'signals.normalized.v1',
        messages: [{ key: sig.signalId, value: JSON.stringify(sig) }],
      });
    } catch {
      await this.redis.del(key); // release the claim so a retry can succeed
      return { ok: false, error: { kind: 'upstream', status: 503, retryAfterMs: 2_000 } };
    }
    return { ok: true, value: sig };
  }
}

// attest/sign.ts — viem 2.x
import { privateKeyToAccount } from 'viem/accounts';
import type { Hex } from 'viem';

const domain = {
  name: 'SignalAttestation',
  version: '1',
  chainId: 1,
  verifyingContract: '0x0000000000000000000000000000000000000001',
} as const;

const types = {
  Signal: [
    { name: 'signalId', type: 'bytes32' },
    { name: 'asset', type: 'string' },
    { name: 'claimType', type: 'string' },
    { name: 'confidenceBps', type: 'uint16' },
    { name: 'observedAt', type: 'uint64' },
  ],
} as const;

export async function attest(
  pk: Hex,
  s: { signalId: Hex; asset: string; claimType: string; confidenceBps: number; observedAtSec: number },
): Promise<{ signature: Hex }> {
  if (!Number.isInteger(s.confidenceBps) || s.confidenceBps < 0 || s.confidenceBps > 10_000) {
    throw new RangeError('confidenceBps must be an integer in [0, 10000]');
  }
  const account = privateKeyToAccount(pk);
  try {
    const signature = await account.signTypedData({
      domain,
      types,
      primaryType: 'Signal',
      message: {
        signalId: s.signalId,
        asset: s.asset,
        claimType: s.claimType,
        confidenceBps: s.confidenceBps,
        observedAt: BigInt(s.observedAtSec),
      },
    });
    return { signature };
  } catch (cause) {
    throw new Error(`attestation failed: ${(cause as Error).message}`, { cause });
  }
}

# risk/gate.py — Python 3.12, Decimal only; no floats anywhere near money
from dataclasses import dataclass
from decimal import Decimal, ROUND_DOWN
from typing import Final

CONFIDENCE_FLOOR: Final = Decimal('0.72')
MAX_NOTIONAL: Final = Decimal('25000')
MAX_PORTFOLIO_FRACTION: Final = Decimal('0.005')  # 50 bps hard cap
MIN_LIQUIDITY: Final = Decimal('2500000')


@dataclass(frozen=True, slots=True)
class Signal:
    signal_id: str
    confidence: Decimal
    verified: bool
    liquidity_usd: Decimal


@dataclass(frozen=True, slots=True)
class Decision:
    approved: bool
    notional: Decimal
    reason: str


def gate(sig: Signal, nav: Decimal) -> Decision:
    if not sig.verified:
        return Decision(False, Decimal('0'), 'unattested_signal')
    if not (Decimal('0') <= sig.confidence <= Decimal('1')):
        return Decision(False, Decimal('0'), 'confidence_out_of_range')
    if sig.confidence < CONFIDENCE_FLOOR:
        return Decision(False, Decimal('0'), 'below_confidence_floor')
    if sig.liquidity_usd < MIN_LIQUIDITY:
        return Decision(False, Decimal('0'), 'insufficient_liquidity')

    raw = min(MAX_NOTIONAL, nav * MAX_PORTFOLIO_FRACTION * sig.confidence)
    notional = raw.quantize(Decimal('0.01'), rounding=ROUND_DOWN)
    if notional <= Decimal('0'):
        return Decision(False, Decimal('0'), 'zero_after_rounding')
    return Decision(True, notional, 'approved')

-- Postgres 16+
CREATE TABLE stake_position (
  position_id      uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  issuer_id        text        NOT NULL,
  numerator_units  numeric(38,0) NOT NULL CHECK (numerator_units > 0),
  denominator_units numeric(38,0) NOT NULL CHECK (denominator_units > 0),
  denominator_asof timestamptz NOT NULL,
  regime           text        NOT NULL CHECK (regime IN ('section_16', 'schedule_13d', 'schedule_13g', 'internal')),
  recorded_at      timestamptz NOT NULL DEFAULT now()
);

-- Derived percentage is never stored mutably; compute on read.
CREATE VIEW stake_pct AS
SELECT position_id, issuer_id, regime,
       (numerator_units / denominator_units)::numeric(12,8) AS pct
FROM stake_position;