viktor/trading
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: productionize local service — fix signal pipeline, lower thresholds, add company-name ticker extraction

Browse source 
- Point Ollama to local instance via host.docker.internal, use gemma3 model
- Remove Docker Ollama service (using host's Ollama instead)
- Add company-name-to-ticker mapping (Apple→AAPL, Tesla→TSLA, etc.) for RSS articles
- Lower signal thresholds for faster feedback with paper trading:
  - FinBERT confidence: 0.6→0.4, signal strength: 0.3→0.15
  - News strategy: article_count 2→1, confidence 0.5→0.3, score ±0.3→±0.15
- Fix market data BarSet access bug (BarSet.__contains__ returns False incorrectly)
- Fix market data SIP feed error by switching to IEX feed for free Alpaca accounts
- Fix nginx proxy routing for /api/auth/* to api-gateway /auth/*
- Add seed_sample_data script
- Update tests for new thresholds and alpaca mock modules
This commit is contained in:
Viktor Barzin 2026-02-22 22:17:26 +00:00
parent 67e64fab18
commit d36ae40df1
No known key found for this signature in database
GPG key ID: 0EB088298288D958
18 changed files with 749 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

12
services/api_gateway/auth/routes.py
View file

@ -167,9 +167,12 @@ async def register_complete(
user_id_str = stored["user_id"]
display_name = stored["display_name"]
# The frontend sends the WebAuthn response under "attestation" or "credential"
credential_data = body.get("credential") or body.get("attestation") or body
try:
verification = verify_registration_response(
credential=body.get("credential", body),
credential=credential_data,
expected_challenge=expected_challenge,
expected_rp_id=config.rp_id,
expected_origin=config.rp_origin,
@ -319,11 +322,14 @@ async def login_complete(
expected_challenge = base64.urlsafe_b64decode(stored["challenge"])
user_id_str = stored["user_id"]
# The frontend sends the WebAuthn response under "assertion" or "credential"
credential_data = body.get("credential") or body.get("assertion") or body
# Look up the credential used
from sqlalchemy import select
from shared.models.auth import UserCredential
credential_id_b64 = body.get("credential", body).get("id", "")
credential_id_b64 = credential_data.get("id", "")
db_session = request.app.state.db_session_factory
async with db_session() as session:
@ -343,7 +349,7 @@ async def login_complete(
try:
verification = verify_authentication_response(
credential=body.get("credential", body),
credential=credential_data,
expected_challenge=expected_challenge,
expected_rp_id=config.rp_id,
expected_origin=config.rp_origin,

58
services/api_gateway/routes/news.py
View file

@ -20,10 +20,13 @@ async def list_news(
max_score: float | None = Query(default=None, ge=-1.0, le=1.0),
page: int = Query(default=1, ge=1),
per_page: int = Query(default=20, ge=1, le=100),
page_size: int | None = Query(default=None, ge=1, le=100),
) -> dict:
"""Recent scored articles with optional filters."""
from shared.models.news import Article, ArticleSentiment
effective_per_page = page_size if page_size is not None else per_page
db = request.app.state.db_session_factory
async with db() as session:
# Base query joining articles with sentiments
@ -54,34 +57,35 @@ async def list_news(
count_query = count_query.where(ArticleSentiment.score <= max_score)
total = (await session.execute(count_query)).scalar() or 0
offset = (page - 1) * per_page
query = query.offset(offset).limit(per_page)
offset = (page - 1) * effective_per_page
query = query.offset(offset).limit(effective_per_page)
result = await session.execute(query)
rows = result.all()
return {
"articles": [
{
"id": str(article.id),
"source": article.source,
"url": article.url,
"title": article.title,
"published_at": (
article.published_at.isoformat()
if article.published_at
else None
),
"fetched_at": article.fetched_at.isoformat(),
"ticker": sentiment.ticker,
"sentiment_score": sentiment.score,
"confidence": sentiment.confidence,
"model_used": sentiment.model_used,
}
for article, sentiment in rows
],
"total": total,
"page": page,
"per_page": per_page,
"pages": (total + per_page - 1) // per_page if per_page else 0,
}
return {
"articles": [
{
"id": str(article.id),
"source": article.source,
"url": article.url,
"title": article.title,
"published_at": (
article.published_at.isoformat()
if article.published_at
else None
),
"fetched_at": article.fetched_at.isoformat(),
"ticker": sentiment.ticker,
"sentiment_score": sentiment.score,
"confidence": sentiment.confidence,
"model_used": sentiment.model_used,
}
for article, sentiment in rows
],
"total": total,
"page": page,
"page_size": effective_per_page,
"per_page": effective_per_page,
"pages": (total + effective_per_page - 1) // effective_per_page if effective_per_page else 0,
}

134
services/api_gateway/routes/portfolio.py
View file

@ -65,33 +65,33 @@ async def get_portfolio(
)
).scalar_one_or_none()
if latest is None:
if latest is None:
return {
"total_value": 0.0,
"cash": 0.0,
"buying_power": 0.0,
"daily_pnl": 0.0,
"daily_pnl_pct": 0.0,
"total_pnl": 0.0,
"total_pnl_pct": 0.0,
"trading_active": True,
}
# Compute percentage fields from snapshot data
daily_pnl_pct = (latest.daily_pnl / (latest.total_value - latest.daily_pnl) * 100.0
if latest.total_value != latest.daily_pnl else 0.0)
return {
"total_value": 0.0,
"cash": 0.0,
"buying_power": 0.0,
"daily_pnl": 0.0,
"daily_pnl_pct": 0.0,
"total_pnl": 0.0,
"total_pnl_pct": 0.0,
"trading_active": True,
"total_value": latest.total_value,
"cash": latest.cash,
"buying_power": latest.cash,
"daily_pnl": latest.daily_pnl,
"daily_pnl_pct": round(daily_pnl_pct, 2),
"total_pnl": latest.daily_pnl, # TODO: compute cumulative P&L from first snapshot
"total_pnl_pct": round(daily_pnl_pct, 2),
"trading_active": True, # TODO: read from Redis trading pause flag
}
# Compute percentage fields from snapshot data
daily_pnl_pct = (latest.daily_pnl / (latest.total_value - latest.daily_pnl) * 100.0
if latest.total_value != latest.daily_pnl else 0.0)
return {
"total_value": latest.total_value,
"cash": latest.cash,
"buying_power": latest.cash,
"daily_pnl": latest.daily_pnl,
"daily_pnl_pct": round(daily_pnl_pct, 2),
"total_pnl": latest.daily_pnl, # TODO: compute cumulative P&L from first snapshot
"total_pnl_pct": round(daily_pnl_pct, 2),
"trading_active": True, # TODO: read from Redis trading pause flag
}
@router.get("/positions")
async def get_positions(
@ -106,18 +106,24 @@ async def get_positions(
result = await session.execute(select(Position))
positions = result.scalars().all()
return [
{
"id": str(p.id),
"ticker": p.ticker,
"qty": p.qty,
"avg_entry": p.avg_entry,
"unrealized_pnl": p.unrealized_pnl or 0.0,
"stop_loss": p.stop_loss,
"take_profit": p.take_profit,
}
for p in positions
]
return [
{
"id": str(p.id),
"ticker": p.ticker,
"qty": p.qty,
"avg_entry": p.avg_entry,
"current_price": round(
p.avg_entry + (p.unrealized_pnl or 0.0) / p.qty, 2
) if p.qty else p.avg_entry,
"unrealized_pnl": p.unrealized_pnl or 0.0,
"unrealized_pnl_pct": round(
(p.unrealized_pnl or 0.0) / (p.avg_entry * p.qty) * 100.0, 2
) if p.avg_entry and p.qty else 0.0,
"stop_loss": p.stop_loss,
"take_profit": p.take_profit,
}
for p in positions
]
@router.get("/metrics")
@ -145,26 +151,26 @@ async def get_portfolio_metrics(
)
strategy_metrics = metrics_result.scalars().all()
total_trades = len(trades)
winning = sum(1 for t in trades if t.pnl is not None and t.pnl > 0)
win_rate = winning / total_trades if total_trades > 0 else 0.0
total_trades = len(trades)
winning = sum(1 for t in trades if t.pnl is not None and t.pnl > 0)
win_rate = winning / total_trades if total_trades > 0 else 0.0
total_pnl = sum(t.pnl for t in trades if t.pnl is not None)
# Approximate ROI from P&L (rough — proper calculation needs initial capital)
roi = total_pnl / 100_000.0 * 100.0 # assumes 100k starting capital
total_pnl = sum(t.pnl for t in trades if t.pnl is not None)
# Approximate ROI from P&L (rough — proper calculation needs initial capital)
roi = total_pnl / 100_000.0 * 100.0 # assumes 100k starting capital
# Average Sharpe from strategy metrics
sharpe_values = [m.sharpe_ratio for m in strategy_metrics if m.sharpe_ratio is not None]
avg_sharpe = sum(sharpe_values) / len(sharpe_values) if sharpe_values else 0.0
# Average Sharpe from strategy metrics
sharpe_values = [m.sharpe_ratio for m in strategy_metrics if m.sharpe_ratio is not None]
avg_sharpe = sum(sharpe_values) / len(sharpe_values) if sharpe_values else 0.0
return {
"roi": round(roi, 4),
"sharpe": round(avg_sharpe, 2),
"win_rate": round(win_rate, 4),
"max_drawdown": 0.0, # TODO: compute from portfolio snapshots
"total_trades": total_trades,
"avg_hold_duration": "0h", # TODO: compute from trade outcomes
}
return {
"roi": round(roi, 4),
"sharpe": round(avg_sharpe, 2),
"win_rate": round(win_rate, 4),
"max_drawdown": 0.0, # TODO: compute from portfolio snapshots
"total_trades": total_trades,
"avg_hold_duration": "0h", # TODO: compute from trade outcomes
}
@router.get("/history")
@ -186,14 +192,14 @@ async def get_portfolio_history(
)
snapshots = result.scalars().all()
return [
{
"timestamp": s.timestamp.isoformat(),
"value": s.total_value,
"total_value": s.total_value,
"cash": s.cash,
"positions_value": s.positions_value,
"daily_pnl": s.daily_pnl,
}
for s in snapshots
]
return [
{
"timestamp": s.timestamp.isoformat(),
"value": s.total_value,
"total_value": s.total_value,
"cash": s.cash,
"positions_value": s.positions_value,
"daily_pnl": s.daily_pnl,
}
for s in snapshots
]

126
services/api_gateway/routes/strategies.py
View file

@ -7,7 +7,7 @@ from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Request, status
from services.api_gateway.auth.middleware import get_current_user
from sqlalchemy import select, desc
from sqlalchemy import select, desc, func
router = APIRouter(prefix="/api/strategies", tags=["strategies"])
@ -17,25 +17,83 @@ async def list_strategies(
request: Request,
_user: dict = Depends(get_current_user),
) -> list[dict]:
"""All strategies with current weights."""
from shared.models.trading import Strategy
"""All strategies with current weights and computed performance fields."""
from shared.models.trading import Strategy, Trade, TradeStatus
db = request.app.state.db_session_factory
async with db() as session:
result = await session.execute(select(Strategy))
strategies = result.scalars().all()
return [
{
"id": str(s.id),
"name": s.name,
"description": s.description,
"current_weight": s.current_weight,
"active": s.active,
"created_at": s.created_at.isoformat() if s.created_at else None,
}
for s in strategies
]
# Compute per-strategy stats from trades table
strategy_stats: dict[UUID, dict] = {}
for s in strategies:
trades_result = await session.execute(
select(Trade).where(
Trade.strategy_id == s.id,
Trade.status == TradeStatus.FILLED,
)
)
trades = trades_result.scalars().all()
total_trades = len(trades)
winning = sum(1 for t in trades if t.pnl is not None and t.pnl > 0)
total_pnl = sum(t.pnl for t in trades if t.pnl is not None)
win_rate = winning / total_trades if total_trades > 0 else 0.0
strategy_stats[s.id] = {
"win_rate": round(win_rate, 4),
"total_pnl": round(total_pnl, 2),
"total_trades": total_trades,
}
return [
{
"id": str(s.id),
"name": s.name,
"description": s.description,
"current_weight": s.current_weight,
"active": s.active,
"win_rate": strategy_stats[s.id]["win_rate"],
"total_pnl": strategy_stats[s.id]["total_pnl"],
"total_trades": strategy_stats[s.id]["total_trades"],
"created_at": s.created_at.isoformat() if s.created_at else None,
}
for s in strategies
]
@router.get("/weight-history")
async def get_all_weight_history(
request: Request,
_user: dict = Depends(get_current_user),
) -> list[dict]:
"""Aggregated weight history pivoted by timestamp for chart display.
Returns data in the format:
``[{"timestamp": "...", "momentum": 0.35, "mean_reversion": 0.30, ...}, ...]``
"""
from shared.models.trading import StrategyWeightHistory, Strategy
db = request.app.state.db_session_factory
async with db() as session:
result = await session.execute(
select(StrategyWeightHistory, Strategy.name)
.join(Strategy, StrategyWeightHistory.strategy_id == Strategy.id)
.order_by(StrategyWeightHistory.created_at)
.limit(200)
)
rows = result.all()
# Pivot: group by timestamp, create one object per timestamp
# with strategy names as keys and new_weight as values
from collections import OrderedDict
pivoted: OrderedDict[str, dict] = OrderedDict()
for h, name in rows:
ts = h.created_at.isoformat() if h.created_at else ""
if ts not in pivoted:
pivoted[ts] = {"timestamp": ts}
pivoted[ts][name] = h.new_weight
return list(pivoted.values())
@router.get("/{strategy_id}/history")
@ -68,16 +126,16 @@ async def get_strategy_weight_history(
)
history = result.scalars().all()
return [
{
"id": str(h.id),
"old_weight": h.old_weight,
"new_weight": h.new_weight,
"reason": h.reason,
"created_at": h.created_at.isoformat() if h.created_at else None,
}
for h in history
]
return [
{
"id": str(h.id),
"old_weight": h.old_weight,
"new_weight": h.new_weight,
"reason": h.reason,
"created_at": h.created_at.isoformat() if h.created_at else None,
}
for h in history
]
@router.get("/{strategy_id}/metrics")
@ -99,13 +157,13 @@ async def get_strategy_metrics(
)
metrics = result.scalars().all()
return [
{
"timestamp": m.timestamp.isoformat(),
"win_rate": m.win_rate,
"total_pnl": m.total_pnl,
"trade_count": m.trade_count,
"sharpe_ratio": m.sharpe_ratio,
}
for m in metrics
]
return [
{
"timestamp": m.timestamp.isoformat(),
"win_rate": m.win_rate,
"total_pnl": m.total_pnl,
"trade_count": m.trade_count,
"sharpe_ratio": m.sharpe_ratio,
}
for m in metrics
]

46
services/api_gateway/routes/trades.py
View file

@ -20,34 +20,44 @@ async def list_trades(
ticker: str | None = Query(default=None),
start_date: datetime | None = Query(default=None),
end_date: datetime | None = Query(default=None),
date_from: datetime | None = Query(default=None),
date_to: datetime | None = Query(default=None),
strategy: str | None = Query(default=None),
profitable: bool | None = Query(default=None),
page: int = Query(default=1, ge=1),
per_page: int = Query(default=20, ge=1, le=100),
page_size: int | None = Query(default=None, ge=1, le=100),
) -> dict:
"""Paginated trade history with optional filters."""
from shared.models.trading import Trade, Strategy
# Accept both parameter naming conventions
effective_per_page = page_size if page_size is not None else per_page
effective_start = start_date or date_from
effective_end = end_date or date_to
db = request.app.state.db_session_factory
async with db() as session:
query = select(Trade).order_by(desc(Trade.created_at))
query = (
select(Trade, Strategy.name.label("strategy_name"))
.outerjoin(Strategy, Trade.strategy_id == Strategy.id)
.order_by(desc(Trade.created_at))
)
count_query = select(func.count()).select_from(Trade)
# Apply filters
if ticker:
query = query.where(Trade.ticker == ticker.upper())
count_query = count_query.where(Trade.ticker == ticker.upper())
if start_date:
query = query.where(Trade.created_at >= start_date)
count_query = count_query.where(Trade.created_at >= start_date)
if end_date:
query = query.where(Trade.created_at <= end_date)
count_query = count_query.where(Trade.created_at <= end_date)
if effective_start:
query = query.where(Trade.created_at >= effective_start)
count_query = count_query.where(Trade.created_at >= effective_start)
if effective_end:
query = query.where(Trade.created_at <= effective_end)
count_query = count_query.where(Trade.created_at <= effective_end)
if strategy:
# Join with Strategy to filter by name
query = query.join(Strategy, Trade.strategy_id == Strategy.id).where(
Strategy.name == strategy
)
# Filter by strategy name (already joined)
query = query.where(Strategy.name == strategy)
count_query = count_query.join(
Strategy, Trade.strategy_id == Strategy.id
).where(Strategy.name == strategy)
@ -61,11 +71,11 @@ async def list_trades(
# Pagination
total = (await session.execute(count_query)).scalar() or 0
offset = (page - 1) * per_page
query = query.offset(offset).limit(per_page)
offset = (page - 1) * effective_per_page
query = query.offset(offset).limit(effective_per_page)
result = await session.execute(query)
trades = result.scalars().all()
rows = result.all()
return {
"trades": [
@ -78,15 +88,17 @@ async def list_trades(
"status": t.status.value,
"pnl": t.pnl,
"strategy_id": str(t.strategy_id) if t.strategy_id else None,
"strategy_name": strategy_name,
"signal_id": str(t.signal_id) if t.signal_id else None,
"created_at": t.created_at.isoformat() if t.created_at else None,
}
for t in trades
for t, strategy_name in rows
],
"total": total,
"page": page,
"per_page": per_page,
"pages": (total + per_page - 1) // per_page if per_page else 0,
"page_size": effective_per_page,
"per_page": effective_per_page,
"pages": (total + effective_per_page - 1) // effective_per_page if effective_per_page else 0,
}

14
services/market_data/main.py
View file

@ -72,6 +72,7 @@ async def _fetch_historical_bars(
Returns the total number of bars published.
"""
from alpaca.data.enums import DataFeed
from alpaca.data.requests import StockBarsRequest
total_published = 0
@ -86,10 +87,14 @@ async def _fetch_historical_bars(
timeframe=timeframe,
start=start,
limit=limit,
feed=DataFeed.IEX,
)
bars = await asyncio.to_thread(client.get_stock_bars, request)
ticker_bars = bars[ticker] if ticker in bars else []
try:
ticker_bars = bars[ticker]
except (KeyError, IndexError):
ticker_bars = []
for bar in ticker_bars:
msg = _bar_to_dict(ticker, bar)
await publisher.publish(msg)
@ -120,6 +125,7 @@ async def _poll_latest_bars(
Returns the number of bars published.
"""
from alpaca.data.enums import DataFeed
from alpaca.data.requests import StockBarsRequest
published = 0
@ -134,10 +140,14 @@ async def _poll_latest_bars(
timeframe=timeframe,
start=start,
limit=1,
feed=DataFeed.IEX,
)
bars = await asyncio.to_thread(client.get_stock_bars, request)
ticker_bars = bars[ticker] if ticker in bars else []
try:
ticker_bars = bars[ticker]
except (KeyError, IndexError):
ticker_bars = []
if ticker_bars:
# Publish only the most recent bar
bar = ticker_bars[-1]

11
services/sentiment_analyzer/analyzers/finbert.py
View file

@ -34,7 +34,7 @@ class FinBERTAnalyzer:
self._pipeline = pipeline(
"sentiment-analysis",
model=self.model_name,
return_all_scores=True,
top_k=None,
)
logger.info("FinBERT model loaded successfully")
return self._pipeline
@ -84,8 +84,9 @@ class FinBERTAnalyzer:
def _parse_scores(results: list[list[dict[str, Any]]]) -> tuple[float, float]:
"""Map pipeline output to ``(score, confidence)``.
The ``return_all_scores=True`` pipeline returns a list of lists of dicts:
``[[{"label": "positive", "score": 0.85}, ...]]``.
With ``top_k=None`` the pipeline returns either:
- ``[[{"label": "positive", "score": 0.85}, ...]]`` (older transformers)
- ``[{"label": "positive", "score": 0.85}, ...]`` (newer transformers)
Mapping:
- ``"positive"`` -> +1
@ -98,8 +99,8 @@ class FinBERTAnalyzer:
"""
label_map = {"positive": 1.0, "negative": -1.0, "neutral": 0.0}
# results is [[{label, score}, ...]]
scores = results[0]
# Handle both [[{label, score}, ...]] and [{label, score}, ...]
scores = results[0] if isinstance(results[0], list) else results
sentiment_score = 0.0
confidence = 0.0

4
services/sentiment_analyzer/config.py
View file

@ -7,8 +7,8 @@ class SentimentAnalyzerConfig(BaseConfig):
"""Extends BaseConfig with sentiment-analysis-specific settings."""
finbert_model: str = "ProsusAI/finbert"
finbert_confidence_threshold: float = 0.6
ollama_model: str = "mistral"
finbert_confidence_threshold: float = 0.4
ollama_model: str = "gemma3"
ollama_host: str = "http://localhost:11434"
max_content_length: int = 512

56
services/sentiment_analyzer/ticker_extractor.py
View file

@ -4,6 +4,7 @@ Handles common formats:
- Dollar-prefixed: ``$AAPL``
- Exchange-prefixed: ``NASDAQ:AAPL``, ``NYSE:TSLA``
- Standalone uppercase words that look like tickers (1-5 uppercase letters)
- Company name mentions: ``Apple``, ``Tesla``, ``Nvidia``, etc.
"""
from __future__ import annotations
@ -119,6 +120,54 @@ _FALSE_POSITIVES: frozenset[str] = frozenset(
}
)
# Mapping of company names (lowercase) to their ticker symbols.
# Longer names are checked first to avoid partial matches.
_COMPANY_TO_TICKER: dict[str, str] = {
"alphabet": "GOOGL",
"google": "GOOGL",
"amazon": "AMZN",
"apple": "AAPL",
"microsoft": "MSFT",
"tesla": "TSLA",
"nvidia": "NVDA",
"meta platforms": "META",
"meta": "META",
"netflix": "NFLX",
"advanced micro devices": "AMD",
"amd": "AMD",
"intel": "INTC",
"broadcom": "AVGO",
"salesforce": "CRM",
"adobe": "ADBE",
"paypal": "PYPL",
"uber": "UBER",
"airbnb": "ABNB",
"spotify": "SPOT",
"shopify": "SHOP",
"snowflake": "SNOW",
"palantir": "PLTR",
"coinbase": "COIN",
"robinhood": "HOOD",
"walmart": "WMT",
"costco": "COST",
"jpmorgan": "JPM",
"goldman sachs": "GS",
"bank of america": "BAC",
"berkshire hathaway": "BRK.B",
"johnson & johnson": "JNJ",
"procter & gamble": "PG",
"coca-cola": "KO",
"disney": "DIS",
"boeing": "BA",
}
# Build a regex that matches any company name as a whole word (case-insensitive).
# Sort by length descending so multi-word names match before single-word subsets.
_COMPANY_PATTERN = re.compile(
r"\b(" + "|".join(re.escape(name) for name in sorted(_COMPANY_TO_TICKER, key=len, reverse=True)) + r")\b",
re.IGNORECASE,
)
# Pattern 1: $AAPL (dollar-sign prefix)
_DOLLAR_PATTERN = re.compile(r"\$([A-Z]{1,5})\b")
@ -152,6 +201,13 @@ def extract_tickers(text: str) -> list[str]:
for match in _EXCHANGE_PATTERN.finditer(text):
_add(match.group(1))
# Company name mentions (case-insensitive).
for match in _COMPANY_PATTERN.finditer(text):
company_name = match.group(1).lower()
ticker = _COMPANY_TO_TICKER.get(company_name)
if ticker:
_add(ticker)
# Standalone uppercase words: only include if they look like real tickers
# (not in the false positives list). We restrict to 2-5 chars to reduce
# noise, unless they were already captured by the dollar/exchange patterns.

2
services/signal_generator/config.py
View file

@ -8,7 +8,7 @@ class SignalGeneratorConfig(BaseConfig):
alpaca_api_key: str = ""
alpaca_secret_key: str = ""
signal_strength_threshold: float = 0.3
signal_strength_threshold: float = 0.15
watchlist: list[str] = []
model_config = {"env_prefix": "TRADING_"}