Event-Driven Workers Architecture¶

Redhound's analysis pipeline runs as isolated Temporal workers communicating through Kafka topics. The API process publishes AnalysisTriggerEvent to Kafka and consumes result events; it never runs the pipeline in-process. Temporal Schedules handle all cron via job-worker. data-worker owns price broadcasting and profile updates via Temporal workflows. PriceBroadcastService and StockProfileUpdater still run as asyncio data-hygiene loops inside the API process (live WS price fanout + warm profile cache).

Architecture Overview¶

flowchart TD
    A["POST /sessions\n(FastAPI API)"] -->|"publish AnalysisTriggerEvent\n+ Kafka headers"| B["Kafka\nanalysis.trigger"]
    B --> C["trigger-consumer\n(Kafka → Temporal bridge)"]
    C -->|"start_workflow\nREJECT_DUPLICATE"| D["Temporal\ntrading namespace"]

    D --> E["analyst-worker\n(9 analysts, parallel)"]
    D --> F["langgraph-worker\n(LLM bull/bear debate)"]
    D --> G["result-consumer\n(DB + Redis + Kafka)"]

    E -->|gRPC :50051| H["model-server\n(DeBERTa sentiment)"]
    G -->|"publish AnalysisCompletedEvent\n+ message key + headers"| I["Kafka\nanalysis.completed"]

    J["data-worker\n(trading namespace)"] -->|"PriceBroadcastWorkflow\ncontinue_as_new"| K["Kafka\ndata.prices"]
    J -->|"StockProfileUpdateWorkflow\n@ 15 min"| L["data.profiles"]

    M["job-worker\n(batch namespace)"] -->|"10 cron workflows\nTemporal Schedules"| N["Kafka\njob.events"]

    style D fill:#4B5EAA,color:#fff
    style B fill:#E8A838,color:#fff
    style I fill:#E8A838,color:#fff

Control Plane vs Data Plane¶

Layer	Technology	Owns
Control plane	Temporal (self-hosted, K8s)	Workflow orchestration, retries, durable state, schedules
Data plane	Kafka (KRaft, K8s)	Event streaming, result fanout, audit log
Cache	Redis (existing)	Live prices, analysis results, agent weights
Persistence	PostgreSQL + PgBouncer	Signal storage, debate history, job outcomes

Temporal Namespaces¶

Workers are split across two Temporal namespaces for isolation:

Namespace	Workers	Retention	Purpose
`trading`	analyst-worker, langgraph-worker, data-worker, result-consumer, trigger-consumer	7 days	Live real-time analysis workflows
`batch`	job-worker	3 days	Cron sweeps isolated from live trading queue

Set REDHOUND_TEMPORAL_NAMESPACE on each worker (default: trading). The Kubernetes namespace-init job creates both namespaces automatically on first deploy.

Task Queue → Worker Mapping¶

Task Queue	Worker	Namespace	Notes
`analyst-task-queue`	`analyst-worker`	`trading`	9 analyst activities + aggregator; sentiment via gRPC model-server
`langgraph-task-queue`	`langgraph-worker`	`trading`	LLM bull/bear debate (activities only, no workflow)
`data-task-queue`	`data-worker`	`trading`	`PriceBroadcastWorkflow` + `StockProfileUpdateWorkflow` + fetch activities
`job-task-queue`	`job-worker`	`batch`	10 Temporal Schedule cron workflows
`result-task-queue`	`result-consumer`	`trading`	DB write, Redis cache, Kafka publish, session update (activities only)

trigger-consumer is a plain Kafka consumer — not a Temporal worker. It bridges analysis.trigger → temporal.start_workflow.

Kafka Topics¶

Topic	Partitions	Retention	Producer	Consumer
`analysis.trigger`	20	1 day	FastAPI (`kafka_producer.py`)	trigger-consumer
`analysis.completed`	10	30 days	result-consumer	WebSocket fanout, audit
`analysis.failed`	6	30 days	result-consumer	Alert consumer
`analysis.partial`	6	1 day	result-consumer	Monitoring
`analysis.timed_out`	4	30 days	result-consumer	Alert consumer
`data.prices`	20	1 day	data-worker	Cache invalidation
`data.profiles`	6	7 days	data-worker	Cache invalidation
`job.events`	4	30 days	job-worker	Audit consumer
`catalyst.detected`	4	7 days	job-worker	trigger-consumer
`analysis.trigger.dlt`	4	30 days	trigger-consumer (on failure)	DLT monitor CronJob
`analysis.completed.dlt`	4	30 days	result-consumer (on failure)	DLT monitor CronJob

DLT topics (*.dlt) accumulate messages that failed all processing retries. The K8s DLT monitor CronJob (k8s/dlt-monitor/) alerts when the count exceeds threshold.

Kafka Message Headers¶

Every message published to Kafka carries these headers:

Header	Value	Set by
`correlation_id`	UUID (= API session ID)	API producer, result-consumer
`source_worker`	`"api"` or `"result-consumer"`	Set at publish site
`schema_version`	`"1.0"`	Set at publish site
`content_type`	`"application/json"`	Set at publish site

Use headers in consumers for tracing and routing without deserialising the payload.

Message Idempotency¶

trigger-consumer uses WorkflowIDReusePolicy.REJECT_DUPLICATE with a deterministic workflow ID (analysis:{symbol}:{event_id}). Duplicate Kafka deliveries silently skip re-starting the same workflow — preventing double-analysis on at-least-once delivery.

Data Worker Workflows¶

Unlike other workers that only host activities, data-worker owns two long-running workflows:

PriceBroadcastWorkflow¶

Runs as a perpetual singleton via workflow.continue_as_new(). Each execution polls live prices then immediately continues as a new workflow run, keeping Temporal history bounded.

Condition	Poll interval
Market hours (Mon–Fri 09:30–16:00 ET)	30 s (configurable via `REDHOUND_PRICE_POLL_INTERVAL_MARKET_HOURS_S`)
Off-hours / weekends	300 s (configurable via `REDHOUND_PRICE_POLL_INTERVAL_OFF_HOURS_S`)

The workflow starts automatically when data-worker starts. If it ever crashes, the Temporal Schedule watchdog (price-broadcast-watchdog, every 1 min) restarts it.

StockProfileUpdateWorkflow¶

Triggered on a Temporal Schedule every 15 minutes. Runs run_refresh_all_profiles to refresh fundamental/technical data for all watched symbols.

Cron Jobs (job-worker Schedules)¶

job-worker manages all background cron work via Temporal Schedules. When job-worker starts it automatically registers all 10 schedules in the batch namespace — idempotent, safe to restart.

Schedule table¶

Schedule ID	Workflow	Fires every	What it does
`price-checks`	`PriceCheckWorkflow`	1 min	Checks watched symbols for significant price moves; triggers re-analysis if threshold exceeded
`scanner-checks`	`ScannerCheckWorkflow`	1 min	Runs configured market scanners; detects opportunities
`outcome-tracking`	`OutcomeTrackingWorkflow`	5 min	Tracks open signal outcomes against current prices
`hotlist-monitor`	`HotlistMonitorWorkflow`	15 min	Monitors a curated hotlist of high-interest symbols
`catalyst-scan`	`CatalystScanWorkflow`	30 min	Scans for catalysts (earnings, news spikes); publishes `catalyst.detected` to Kafka
`weight-update`	`WeightUpdateWorkflow`	60 min	Recomputes per-analyst accuracy weights from recent signal outcomes
`index-intraday`	`IndexIntradayWorkflow`	5 min	Refreshes S&P 500 / NASDAQ 100 constituent lists (intraday)
`universe-sweep`	`UniverseSweepWorkflow`	24 h	Full universe analysis sweep across all watched symbols
`action-plan-eval`	`ActionPlanEvalWorkflow`	24 h	Evaluates open action plans against current market conditions
`index-eod`	`IndexEodWorkflow`	24 h	End-of-day index constituent refresh

Starting the cron worker¶

# Via Docker Compose (recommended — Temporal starts automatically)
make up-workers

# As a local process (Temporal must already be running)
make run-job-worker

That's it. No extra setup — schedules are created on first startup.

Triggering a job manually (without waiting for the timer)¶

Use the Temporal CLI or the Web UI:

# Trigger any schedule immediately
temporal schedule trigger --schedule-id price-checks \
  --address localhost:7233 --namespace batch

temporal schedule trigger --schedule-id universe-sweep \
  --address localhost:7233 --namespace batch

Or open the Temporal Web UI (make temporal-ui → localhost:8233), go to Schedules, and click Trigger Now.

Listing and inspecting schedules¶

# List all schedules (shows next fire time and last run status)
temporal schedule list --address localhost:7233 --namespace batch

# Inspect one schedule in detail
temporal schedule describe --schedule-id weight-update \
  --address localhost:7233 --namespace batch

Pausing and resuming a schedule¶

temporal schedule pause --schedule-id universe-sweep \
  --address localhost:7233 --namespace batch \
  --reason "maintenance window"

temporal schedule unpause --schedule-id universe-sweep \
  --address localhost:7233 --namespace batch

Adjusting intervals¶

Intervals come from env vars. Override before starting job-worker:

# Run universe-sweep every 12 h instead of 24 h
REDHOUND_UNIVERSE_SWEEP_CRON=43200 make run-job-worker

Model Server¶

model-server is a dedicated gRPC inference pod. Workers call it via ModelServerClient — no worker loads model weights directly.

flowchart LR
    A["analyst-worker"] -->|"gRPC :50051\nInfer(model_name='deberta-sentiment', text='...')"| B["model-server"]
    B --> C["deberta-sentiment\n(mrm8488/deberta-v3-ft-financial-news-sentiment-analysis)"]
    B -.-> D["future models\n(add to MODEL_MAP)"]

Health: GET /ready returns 200 only after all configured models pass warmup inference.

Add a new model:

Create backend/model_server/models/<name>.py implementing BaseInferenceModel
Register in MODEL_MAP in backend/model_server/models/__init__.py
Add name to REDHOUND_MODEL_SERVER_MODELS env var (comma-separated)

Local Development¶

The worker stack is required for analyses to complete locally — the API publishes triggers to Kafka and the workers run the pipeline.

Docker Compose (recommended)¶

# First run: images are built automatically if missing (takes a few minutes).
# Subsequent runs reuse cached images and start in seconds.
make up-workers    # Temporal + Kafka + all workers + model-server
make logs-workers  # follow all worker logs
make down-workers  # stop

To pre-build images manually (e.g. before a demo or to check for compile errors):

make build-all-workers    # build all 6 worker images
make build-model-server   # build model-server image (bakes DeBERTa weights)

Running individual workers (local process)¶

Start Temporal + Kafka via Docker first, then run any worker directly:

# Prerequisites: infra must be running (workers images are built automatically)
make up-workers   # or start Temporal + Kafka containers individually

# Run each worker in its own terminal
make run-analyst-worker      # analyst-task-queue  (port 8080)
make run-langgraph-worker    # langgraph-task-queue (port 8081)
make run-data-worker         # data-task-queue      (port 8082)
make run-job-worker          # job-task-queue       (port 8083)
make run-result-consumer     # result-task-queue    (port 8084)
make run-trigger-consumer    # Kafka → Temporal bridge (port 8085)
make run-model-server        # gRPC :50051, HTTP :8080

Override any default with environment variables:

REDHOUND_TEMPORAL_ADDR=localhost:7233 \
REDHOUND_KAFKA_BOOTSTRAP=localhost:9092 \
REDHOUND_TEMPORAL_NAMESPACE=trading \
make run-analyst-worker

With workers (Kubernetes)¶

See k8s/README.md in the repository for the full operator runbook.

Quick start:

make k8s-apply-infra   # Kafka + Temporal + PgBouncer (wait ~60s for Temporal init)
make k8s-apply-workers # model-server + all workers + trigger-consumer
make k8s-status        # check pod status
make temporal-ui       # port-forward Temporal Web UI → localhost:8233

Contracts Package¶

All inter-worker interfaces live in backend/contracts/. Workers must import only from here — never from each other's package. Enforced by scripts/check_cross_worker_imports.py (pre-commit hook).

backend/contracts/
├── kafka/topics.py           # Topic name constants
├── kafka/schemas/            # Pydantic event models (frozen/immutable, extra="ignore")
├── temporal/task_queues.py   # Task queue name constants
├── temporal/workflows.py     # Workflow/Activity I/O models
├── grpc/model_client.py      # ModelServerClient gRPC stub wrapper
├── models/signals.py         # Shared AgentSignal, FinalSignal types
└── utils/market_calendar.py  # is_market_open() via pandas-market-calendars

Worker Health Endpoints¶

Every worker exposes a FastAPI health server (default port 8080):

Endpoint	Returns	Notes
`GET /health`	`{"status": "ok", "task_queue": "..."}`	Always 200
`GET /ready`	`{"status": "ready"}`	503 until Temporal connected
`GET /metrics`	Prometheus text format	`activity_duration_seconds`, `activity_retry_count`

Structured Logging¶

Workers emit JSON logs with a shared schema:

{
  "timestamp": "2026-03-27T09:31:04.123Z",
  "level": "INFO",
  "correlation_id": "abc-123",
  "workflow_id": "analysis:AAPL:evt-456",
  "activity": "TechnicalAnalystActivity",
  "worker": "analyst-worker",
  "message": "Activity completed",
  "duration_ms": 342
}

correlation_id is propagated via a Python contextvars.ContextVar populated at activity start and carried through Kafka message headers. Use set_correlation_id(cid) from backend.workers.common.logging at the top of each activity.

Environment Variables¶

Variable	Default	Notes
`REDHOUND_TEMPORAL_ADDR`	`localhost:7233`	Temporal frontend address
`REDHOUND_TEMPORAL_NAMESPACE`	`trading`	`trading` for real-time workers, `batch` for job-worker
`REDHOUND_KAFKA_BOOTSTRAP`	`localhost:9092`	Kafka bootstrap server
`REDHOUND_MODEL_SERVER_ADDR`	`model-server:50051`	gRPC address for model-server
`REDHOUND_MODEL_SERVER_MODELS`	`deberta-sentiment`	Comma-separated model names to load
`REDHOUND_PRICE_POLL_INTERVAL_MARKET_HOURS_S`	`30.0`	PriceBroadcastWorkflow poll rate (market hours)
`REDHOUND_PRICE_POLL_INTERVAL_OFF_HOURS_S`	`300.0`	PriceBroadcastWorkflow poll rate (off-hours)
`REDHOUND_ANALYSIS_WORKFLOW_TIMEOUT_MIN`	`5.0`	Max duration for AnalysisWorkflow
`REDHOUND_LANGGRAPH_TIMEOUT_MIN`	`8.0`	Max duration for LangGraph debate activity

Adding a New Activity¶

Create backend/workers/<worker>/activities/<name>.py:

from temporalio import activity
from backend.contracts.temporal.workflows import MyInput

@activity.defn(name="run_my_activity")
async def run_my_activity(inp: MyInput) -> None:
    ...

Add input/output models to backend/contracts/temporal/workflows.py
Register in backend/workers/<worker>/main.py Worker activities list
Call from workflow: workflow.execute_activity("run_my_activity", inp, ...)
Write unit tests in tests/workers/<worker>/

Schema Evolution¶

Kafka event schemas use Pydantic v2 with model_config = ConfigDict(extra="ignore") — unknown fields from future producers are silently dropped, ensuring forward compatibility.

Rule: New fields MUST have defaults. Non-optional fields without defaults break old consumers.

# Safe — old consumers ignore new_field
class MyEvent(BaseModel):
    model_config = ConfigDict(extra="ignore")
    existing_field: str
    new_field: str = "default"   # has a default

# Breaks old consumers
class MyEvent(BaseModel):
    existing_field: str
    new_field: str               # no default — DO NOT DO THIS