Financial Trading Dashboard — component-framework

The Scenario

A retail trading platform needs real-time portfolio P&L, live price feeds, and order execution — without a JS SPA build chain. Four server-side components handle the entire dashboard:

PortfolioComponent — displays positions, P&L, and total value; updates via WebSocket whenever prices change.
WatchlistComponent — live list of securities with bid/ask/last prices; WebSocket subscription per symbol.
OrderComponent — handles buy/sell order submission with server-side validation and optimistic UI so the button responds instantly.
RiskMetricsComponent — margin usage, exposure by sector, and real-time alerts whenever account limits are breached.

The component-framework answer: write Python. The server owns all positions, P&L calculations, and risk computations. The browser sends events and receives rendered HTML — there is no client-side state machine to maintain or keep in sync.

Architecture

Browser (HTMX + WebSocket) │ ├── POST /components/order/ → OrderComponent.on_place_order() ├── POST /components/watchlist/ → WatchlistComponent.on_add_symbol() ├── GET /components/portfolio/ → PortfolioComponent.mount() │ └── ws://…/ws/ ├── subscribe: "prices" → WatchlistComponent + PortfolioComponent ├── broadcast: order_fill → PortfolioComponent (position update) └── broadcast: risk_breach → RiskMetricsComponent (alert) Price Feed Worker (Celery / background task) └── Every 250ms: broadcast price ticks → ws_manager.broadcast("prices", {...})

Key insight: the server controls all state and all financial computations. The browser renders whatever it receives. There is no client-side P&L arithmetic, no local price cache, and no synchronisation logic anywhere in the JavaScript layer.

PortfolioComponent

Loads the authenticated user's open positions from the database on mount, then updates P&L in real time as WebSocket price ticks arrive — without polling.

@registry.register("portfolio")
class PortfolioComponent(Component):
    template_name = "components/portfolio.html"
    permission_classes = [IsAuthenticated]
    cache_key_prefix = "portfolio"
    cache_ttl = 5  # seconds — avoid DB hammering on tick storms

    def mount(self):
        user = self.request.user
        positions = Position.objects.filter(
            account__user=user, quantity__gt=0
        ).select_related("symbol")
        self.state["positions"] = [
            {"symbol": p.symbol.ticker,
             "qty":    p.quantity,
             "avg_cost": float(p.avg_cost),
             "last":   float(p.symbol.last_price),
             "pnl":    float((p.symbol.last_price - p.avg_cost) * p.quantity)}
            for p in positions
        ]
        self.state["total_value"] = sum(
            p["last"] * p["qty"] for p in self.state["positions"]
        )

    async def on_ws_message(self, message: dict):
        if message.get("event") == "price_tick":
            ticks = message["prices"]  # {ticker: last_price}
            for pos in self.state["positions"]:
                if pos["symbol"] in ticks:
                    pos["last"] = ticks[pos["symbol"]]
                    pos["pnl"]  = (pos["last"] - pos["avg_cost"]) * pos["qty"]
            self.state["total_value"] = sum(
                p["last"] * p["qty"] for p in self.state["positions"]
            )

The template renders a positions table with color-coded P&L — green for gains, rose for losses — and a total value row at the foot of the table.

<div id="portfolio-{{ component_id }}" hx-target="this" hx-swap="outerHTML">
  <table class="positions-table">
    <thead><tr>
      <th>Symbol</th><th>Qty</th><th>Avg Cost</th><th>Last</th><th>P&amp;L</th>
    </tr></thead>
    <tbody>
      {% for pos in state.positions %}
      <tr>
        <td class="ticker">{{ pos.symbol }}</td>
        <td>{{ pos.qty }}</td>
        <td>{{ pos.avg_cost|floatformat:2 }}</td>
        <td>{{ pos.last|floatformat:2 }}</td>
        <td class="{% if pos.pnl >= 0 %}pnl-pos{% else %}pnl-neg{% endif %}">
          {{ pos.pnl|floatformat:2 }}
        </td>
      </tr>
      {% endfor %}
    </tbody>
    <tfoot><tr>
      <td colspan="4">Total Value</td>
      <td class="total-value">{{ state.total_value|floatformat:2 }}</td>
    </tr></tfoot>
  </table>
</div>

OrderComponent with Optimistic UI

Handles buy and sell order submission. Pydantic validates the payload before any database write. Optimistic UI gives instant button feedback while the order travels to the exchange routing layer.

from typing import Annotated, Literal
from pydantic import BaseModel, Field


class OrderSchema(BaseModel):
    symbol:     str
    side:       Literal["buy", "sell"]
    quantity:   Annotated[int, Field(gt=0, le=10_000)]
    order_type: Literal["market", "limit"] = "market"
    limit_price: float | None = None


@registry.register("order")
class OrderComponent(FormComponent):
    schema = OrderSchema
    template_name = "components/order.html"
    permission_classes = [IsAuthenticated, HasTradingPermission]
    rate_limit = "60/minute"  # comply with exchange rules

    def get_optimistic_patch(self, event: str, payload: dict) -> dict:
        # Instant UI feedback before server round-trip completes
        if event == "submit":
            return {"status": "pending", "message": "Sending order…"}
        return {}

    def on_submit(self):
        order = Order.objects.create(
            account=self.request.user.account,
            symbol=self.validated_data["symbol"],
            side=self.validated_data["side"],
            quantity=self.validated_data["quantity"],
            order_type=self.validated_data["order_type"],
            limit_price=self.validated_data.get("limit_price"),
        )
        self.state["status"]   = "submitted"
        self.state["order_id"] = order.pk
        self.state["message"]  = f"Order #{order.pk} submitted"

Rate limiting: rate_limit = "60/minute" prevents accidental order storms and satisfies typical exchange throttle requirements. Requests beyond the limit receive HTTP 429 with a Retry-After header; the client rolls back any optimistic state automatically.

Optimistic UI: get_optimistic_patch() returns {"status": "pending", "message": "Sending order…"} instantly — before the server confirms — exactly like native trading apps. If the server returns a validation error, the client rolls back to the pre-patch state automatically.

WatchlistComponent with Live Prices

Renders a tight bid/ask/last data table that flashes price cells on change. Users can add symbols interactively; WebSocket ticks keep every row current.

@registry.register("watchlist")
class WatchlistComponent(Component):
    template_name = "components/watchlist.html"
    permission_classes = [IsAuthenticated]

    def mount(self):
        items = WatchlistItem.objects.filter(
            user=self.request.user
        ).select_related("symbol").order_by("rank")
        self.state["items"] = [
            {"ticker":  i.symbol.ticker,
             "name":    i.symbol.name,
             "last":    float(i.symbol.last_price),
             "bid":     float(i.symbol.bid),
             "ask":     float(i.symbol.ask),
             "chg_pct": float(i.symbol.change_pct)}
            for i in items
        ]

    def on_add_symbol(self, ticker: str):
        symbol = Symbol.objects.get(ticker=ticker.upper())
        WatchlistItem.objects.get_or_create(
            user=self.request.user, symbol=symbol,
            defaults={"rank": len(self.state["items"])}
        )
        self.state["items"].append({
            "ticker":  symbol.ticker,
            "name":    symbol.name,
            "last":    float(symbol.last_price),
            "bid":     float(symbol.bid),
            "ask":     float(symbol.ask),
            "chg_pct": float(symbol.change_pct),
        })

    async def on_ws_message(self, message: dict):
        if message.get("event") == "price_tick":
            ticks = message["prices"]
            for item in self.state["items"]:
                if item["ticker"] in ticks:
                    item["last"] = ticks[item["ticker"]]

The template renders a compact data table with flashing price cells. A CSS transition on the last cell triggers a brief highlight whenever the value changes — no JavaScript animation library required.

<div id="watchlist-{{ component_id }}">
  <form
    hx-post="{% url 'watchlist_component' %}"
    hx-vals='{"event": "add_symbol"}'
    hx-target="#watchlist-{{ component_id }}"
    hx-swap="outerHTML">
    <input name="ticker" placeholder="Add symbol…" autocomplete="off" />
    <button type="submit">+</button>
  </form>
  <table class="watchlist-table">
    <thead><tr>
      <th>Ticker</th><th>Bid</th><th>Ask</th><th>Last</th><th>Chg%</th>
    </tr></thead>
    <tbody>
      {% for item in state.items %}
      <tr>
        <td class="ticker">{{ item.ticker }}</td>
        <td>{{ item.bid|floatformat:2 }}</td>
        <td>{{ item.ask|floatformat:2 }}</td>
        <td class="last-price">{{ item.last|floatformat:2 }}</td>
        <td class="{% if item.chg_pct >= 0 %}chg-pos{% else %}chg-neg{% endif %}">
          {{ item.chg_pct|floatformat:2 }}%
        </td>
      </tr>
      {% endfor %}
    </tbody>
  </table>
</div>

RiskMetricsComponent

Displays margin utilisation, sector exposure, and a live alert feed. The component subscribes to a per-account WebSocket channel; the background worker broadcasts a risk_breach event whenever utilisation exceeds a threshold.

@registry.register("risk_metrics")
class RiskMetricsComponent(Component):
    template_name = "components/risk_metrics.html"
    permission_classes = [IsAuthenticated]
    cache_key_prefix = "risk"
    cache_ttl = 10

    def mount(self):
        account = self.request.user.account
        self.state["margin_used"]     = float(account.margin_used)
        self.state["margin_limit"]    = float(account.margin_limit)
        self.state["utilisation_pct"] = self._pct(account)
        self.state["alerts"]          = []
        self.state["sector_exposure"]  = self._sector_breakdown()

    def _pct(self, account) -> float:
        if account.margin_limit == 0:
            return 0.0
        return round(float(account.margin_used / account.margin_limit * 100), 1)

    async def on_ws_message(self, message: dict):
        if message.get("event") == "risk_breach":
            self.state["alerts"].append({
                "level": message["level"],  # "warn" | "critical"
                "text":  message["text"],
                "ts":    message["ts"],
            })

The template renders a margin gauge (an HTML progress bar) and a scrollable alert feed below it.

<div id="risk-{{ component_id }}">
  <h3>Margin Utilisation</h3>
  <div class="gauge">
    <div
      class="gauge__fill {% if state.utilisation_pct >= 95 %}gauge--critical{% elif state.utilisation_pct >= 80 %}gauge--warn{% endif %}"
      style="width: {{ state.utilisation_pct }}%">
    </div>
  </div>
  <p>{{ state.margin_used|floatformat:0 }} / {{ state.margin_limit|floatformat:0 }}
     ({{ state.utilisation_pct }}%)</p>

  {% if state.alerts %}
  <ul class="alert-feed">
    {% for alert in state.alerts %}
    <li class="alert alert--{{ alert.level }}">
      <span class="alert__ts">{{ alert.ts }}</span>
      {{ alert.text }}
    </li>
    {% endfor %}
  </ul>
  {% endif %}
</div>

Push, not poll: when margin utilisation exceeds 80%, ws_manager broadcasts a risk_breach event to the account's private channel. The component appends a critical alert row without any polling loop. The browser receives the update within one WebSocket frame.

Background Price Feed

A Celery periodic task fetches the latest market data and pushes it into the WebSocket manager every 250 ms. No component polls — the push model means the browser is only updated when data actually changes.

from datetime import datetime
from celery import shared_task
from component_framework.core import ws_manager
from .models import Symbol, Account


@shared_task
async def broadcast_price_ticks():
    """Fetch latest prices and push to all subscribed components."""
    tickers = Symbol.objects.values_list("ticker", flat=True)
    prices = await fetch_prices(tickers)  # external market data API

    await ws_manager.broadcast("prices", {
        "event":  "price_tick",
        "prices": prices,  # {AAPL: 189.34, MSFT: 415.22, …}
    })

    # Check per-account risk limits and alert when breached
    for account in Account.objects.annotate_utilisation():
        if account.utilisation_pct > 80:
            level = "warn" if account.utilisation_pct < 95 else "critical"
            await ws_manager.broadcast(f"risk-{account.user_id}", {
                "event": "risk_breach",
                "level": level,
                "text":  f"Margin at {account.utilisation_pct:.1f}%",
                "ts":    datetime.utcnow().isoformat(),
            })

Push, not poll: no component polls a database or an API on a timer. The background task pushes ticks at 250 ms; components re-render only the rows that changed. Under a tick storm the cache_ttl guard on PortfolioComponent prevents redundant DB reads.

Testing

Every component is pure Python. The test suite runs at unit-test speed — no broker connections, no market data subscriptions, no running Django server.

import asyncio
from django.contrib.auth.models import AnonymousUser
from component_framework.testing import (
    ComponentTestCase, dispatch_event, assert_state, assert_permission_denied
)
from trading.components import OrderComponent, PortfolioComponent


class TestOrderComponent(ComponentTestCase):

    def test_valid_market_order_is_submitted(self):
        component = dispatch_event(
            OrderComponent, "submit",
            {"symbol": "AAPL", "side": "buy",
             "quantity": 10, "order_type": "market"},
        )
        assert component.state["status"] == "submitted"
        assert Order.objects.filter(symbol="AAPL").count() == 1

    def test_quantity_zero_is_rejected(self):
        component = dispatch_event(
            OrderComponent, "submit",
            {"symbol": "AAPL", "side": "buy",
             "quantity": 0, "order_type": "market"},
        )
        assert "quantity" in component.state["errors"]

    def test_unauthenticated_cannot_place_order(self):
        assert_permission_denied(
            OrderComponent, "submit", user=AnonymousUser()
        )

    def test_optimistic_patch_shows_pending(self):
        component = OrderComponent()
        patch = component.get_optimistic_patch("submit", {})
        assert patch["status"] == "pending"

    def test_price_tick_updates_portfolio_pnl(self):
        component = PortfolioComponent()
        component.state = {
            "positions": [{
                "symbol": "AAPL", "qty": 10,
                "avg_cost": 180.0, "last": 180.0, "pnl": 0,
            }],
            "total_value": 1800,
        }
        asyncio.run(component.on_ws_message({
            "event":  "price_tick",
            "prices": {"AAPL": 195.0},
        }))
        assert component.state["positions"][0]["pnl"] == 150.0
        assert component.state["total_value"] == 1950.0

Unit-test speed: all component logic is pure Python — tests run without broker connections or market data subscriptions. The WebSocket message handler is a plain async def called with asyncio.run(); no mock servers or test channels are needed.

Summary

Component	Key Feature	WebSocket	Rate-Limited
PortfolioComponent	Real-time P&L	✓ subscribe	—
WatchlistComponent	Live bid/ask	✓ subscribe	—
OrderComponent	Optimistic UI	—	✓ 60/min
RiskMetricsComponent	Margin alerts	✓ subscribe	—

The complete trading dashboard — live price feeds, P&L updates, order entry with optimistic UI, and real-time risk alerts — is built with four Python classes and four Django templates. No JavaScript framework. No build toolchain. No client-side state machine. Just Python.