Skip to main content

mts1b_foundation.signals — full reference

What strategy output looks like on the wire.

Signal​

class Signal(BaseModel):
    signal_id: str
    fund_id: str
    asof: datetime
    factor_name: str               # e.g. "f_momentum_12_1"
    params: dict = {}
    universe: list[str] = []
    weights: dict[str, float] = {}
    metadata: dict = {}

A strategy's output at ONE rebalance: target weights + traceability metadata.

Example​

from datetime import datetime, timezone
from mts1b_foundation.signals import Signal

signal = Signal(
    signal_id="sig-2026-05-23-momentum-001",
    fund_id="paper-momentum",
    asof=datetime(2026, 5, 23, 16, 0, tzinfo=timezone.utc),
    factor_name="f_momentum_12_1",
    params={"h_long": 252, "h_skip": 21},
    universe=["AAPL", "MSFT", "GOOG", "META", "NVDA", "AMZN", "TSLA", "JPM"],
    weights={
        "AAPL": 0.18,
        "MSFT": 0.20,
        "GOOG": -0.05,
        "NVDA": 0.25,
        "AMZN": 0.12,
        "TSLA": -0.10,
        "JPM": 0.10,
    },
    metadata={
        "ic": 0.062,
        "t_stat": 6.41,
        "backtest_sharpe_walkforward": 1.43,
        "n_long": 4,
        "n_short": 2,
        "gross_exposure": 1.0,
        "net_exposure": 0.70,
    },
)

Validation rules (enforced upstream, not in pydantic)​

  • weights.keys() ⊆ universe
  • sum(abs(w)) ≈ gross_exposure (from metadata)
  • All weights in [-1, 1]
def validate(s: Signal):
    weight_syms = set(s.weights.keys())
    universe_set = set(s.universe)
    extra = weight_syms - universe_set
    if extra:
        raise ValueError(f"weights for unknown symbols: {extra}")

    gross = sum(abs(w) for w in s.weights.values())
    expected_gross = s.metadata.get("gross_exposure", 1.0)
    if abs(gross - expected_gross) > 0.05:
        raise ValueError(f"gross {gross} vs expected {expected_gross}")

Publishing to NATS​

from mts1b_foundation.nats import subject

subj = subject("mts1b-research", "signals", "published")
await js.publish(
    subject=subj,
    payload=signal.model_dump_json().encode(),
    headers={
        "NATS-Msg-Id": signal.signal_id,
        "mts-schema": "mts1b_foundation.signals.Signal:v1",
        "mts-source-repo": "mts1b-research",
    },
)

Consuming + acting on signals​

async for raw in sub.messages:
    signal = Signal.model_validate_json(raw.data)
    current_positions = await oms.get_positions(fund_id=signal.fund_id)
    nav = await treasury.get_nav(signal.fund_id)

    target_weights = signal.weights
    diffs = compute_diff(current_positions, target_weights, nav)
    for d in diffs:
        await oms.submit(d.to_order(strategy_id=signal.factor_name))
    await raw.ack()

TargetWeight​

class TargetWeight(BaseModel):
    fund_id: str
    symbol: Symbol
    target_qty: Decimal
    target_usd: Decimal
    side: Side
    confidence: float = 1.0

A per-symbol target after sizing. Used internally by mts1b-portfolio and mts1b-oms.routing.

from mts1b_foundation.signals import TargetWeight
from mts1b_foundation.symbology import Symbol
from mts1b_foundation.orders import Side

target = TargetWeight(
    fund_id="paper-momentum",
    symbol=Symbol("AAPL"),
    target_qty=Decimal("100"),
    target_usd=Decimal("18000"),
    side=Side.BUY,
    confidence=0.85,           # factor's confidence — usable by adaptive sizing
)

Confidence weighting​

Some sizers reduce weight when confidence is low:

def confidence_scaled_target(t: TargetWeight, min_conf: float = 0.5) -> TargetWeight:
    if t.confidence < min_conf:
        scale = t.confidence / min_conf
        return t.model_copy(update={
            "target_qty": t.target_qty * Decimal(str(scale)),
            "target_usd": t.target_usd * Decimal(str(scale)),
        })
    return t

Decoding a Signal into orders​

def signal_to_orders(signal: Signal, current_positions: dict[Symbol, Decimal],
                     nav: Decimal, prices: dict[Symbol, Decimal]) -> list[Order]:
    """Convert target weights into orders, given current positions."""
    orders = []
    for sym_str, weight in signal.weights.items():
        sym = Symbol(sym_str)
        target_usd = Decimal(str(weight)) * nav
        target_qty = target_usd / prices[sym]
        current_qty = current_positions.get(sym, Decimal("0"))
        delta = target_qty - current_qty

        if abs(delta) < Decimal("0.01"):   # rounding noise
            continue

        side = Side.BUY if delta > 0 else Side.SELL
        orders.append(Order(
            order_id=f"ord-{uuid.uuid4()}",
            idempotency_key=f"{signal.factor_name}-{signal.asof.isoformat()}-{sym}-{side.value}",
            symbol=sym,
            side=side,
            quantity=abs(delta),
            order_type=OrderType.LIMIT,
            limit_price=prices[sym],
            fund_id=signal.fund_id,
            strategy_id=signal.factor_name,
            broker="paper",
            actor="signal_executor",
            created_at=datetime.now(timezone.utc),
        ))
    return orders

See also​