diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000..fb0a2f8 --- /dev/null +++ b/backend/app/api/__init__.py @@ -0,0 +1 @@ +"""API route modules.""" diff --git a/backend/app/api/chat.py b/backend/app/api/chat.py new file mode 100644 index 0000000..bd1a529 --- /dev/null +++ b/backend/app/api/chat.py @@ -0,0 +1,189 @@ +"""AI chat API endpoint.""" +from __future__ import annotations + +import json +import logging +import os + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from app import db +from app import state + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/chat", tags=["chat"]) + +_LLM_MOCK = os.environ.get("LLM_MOCK", "").lower() == "true" + +_MOCK_RESPONSE = { + "message": ( + "I'm analyzing your portfolio. You have a diversified mix of holdings. " + "Overall looking healthy — would you like me to suggest any rebalancing?" + ), + "trades": [], + "watchlist_changes": [], +} + +_SYSTEM_PROMPT = """\ +You are FinAlly, an AI trading assistant for a simulated stock trading workstation. +Help users analyze their portfolio, suggest trades, and manage their watchlist. + +Always respond with a single JSON object matching exactly this schema: +{ + "message": "", + "trades": [{"ticker": "SYMBOL", "side": "buy|sell", "quantity": }], + "watchlist_changes": [{"ticker": "SYMBOL", "action": "add|remove"}] +} + +Rules: +- Be concise and data-driven. +- Only execute trades when the user explicitly asks. +- Acknowledge this is a simulated portfolio with fake money. +- Respond with valid JSON only — no extra text.""" + + +class ChatRequest(BaseModel): + message: str + + +async def _call_llm(messages: list[dict]) -> dict: + try: + import litellm + + response = await litellm.acompletion( + model="openrouter/openai/gpt-oss-120b", + messages=messages, + api_base="https://openrouter.ai/api/v1", + api_key=os.environ.get("OPENROUTER_API_KEY", ""), + response_format={"type": "json_object"}, + temperature=0.7, + extra_headers={ + "X-Title": "FinAlly", + "HTTP-Referer": "https://github.com/eluiken/finally", + }, + ) + return json.loads(response.choices[0].message.content) + except Exception: + logger.exception("LLM call failed") + return { + "message": "I encountered an error processing your request. Please try again.", + "trades": [], + "watchlist_changes": [], + } + + +@router.post("") +async def chat(req: ChatRequest) -> dict: + """Send a message; receive a response with optional auto-executed trades.""" + if not req.message.strip(): + raise HTTPException(status_code=422, detail="Message cannot be empty") + + portfolio = await db.get_portfolio() + cash: float = portfolio["cash"] + positions: list[dict] = portfolio["positions"] + watchlist_tickers = await db.get_watchlist() + + pos_lines = [] + positions_value = 0.0 + for pos in positions: + ticker = pos["ticker"] + qty = pos["quantity"] + avg = pos["avg_cost"] + cur = state.price_cache.get_price(ticker) or avg + val = qty * cur + pnl = (cur - avg) * qty + positions_value += val + pos_lines.append( + f" {ticker}: {qty:.2f}sh @ avg ${avg:.2f}, now ${cur:.2f}, P&L ${pnl:+.2f}" + ) + + wl_lines = [] + for t in watchlist_tickers: + u = state.price_cache.get(t) + wl_lines.append(f"{t}:${u.price:.2f}" if u else f"{t}:N/A") + + context = ( + f"Portfolio: cash=${cash:.2f}, total=${cash + positions_value:.2f}\n" + f"Positions:\n" + ("\n".join(pos_lines) if pos_lines else " (none)") + "\n" + f"Watchlist: {', '.join(wl_lines)}" + ) + + history = await db.get_chat_history(limit=20) + messages: list[dict] = [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": context}, + ] + for msg in history: + messages.append({"role": msg["role"], "content": msg["content"]}) + messages.append({"role": "user", "content": req.message}) + + await db.save_message("user", req.message) + + llm_result = _MOCK_RESPONSE if _LLM_MOCK else await _call_llm(messages) + + response_message = llm_result.get("message", "") + trades = llm_result.get("trades", []) + wl_changes = llm_result.get("watchlist_changes", []) + + executed: list[dict] = [] + errors: list[str] = [] + for trade in trades: + ticker = str(trade.get("ticker", "")).upper() + side = str(trade.get("side", "")).lower() + try: + quantity = float(trade.get("quantity", 0)) + except (TypeError, ValueError): + errors.append(f"Invalid quantity in trade: {trade}") + continue + if not ticker or side not in ("buy", "sell") or quantity <= 0: + errors.append(f"Invalid trade spec: {trade}") + continue + price = state.price_cache.get_price(ticker) + if price is None: + errors.append(f"No price available for {ticker}") + continue + result = await db.execute_trade(ticker, side, quantity, price) + if "error" in result: + errors.append(result["error"]) + else: + executed.append(result) + + applied_wl: list[dict] = [] + for change in wl_changes: + ticker = str(change.get("ticker", "")).upper() + action = str(change.get("action", "")).lower() + if action == "add": + if await db.add_ticker(ticker) and state.market_source: + await state.market_source.add_ticker(ticker) + applied_wl.append({"ticker": ticker, "action": "add"}) + elif action == "remove": + if await db.remove_ticker(ticker): + if state.market_source: + await state.market_source.remove_ticker(ticker) + state.price_cache.remove(ticker) + applied_wl.append({"ticker": ticker, "action": "remove"}) + + actions = None + if executed or errors or applied_wl: + actions = json.dumps({"trades": executed, "errors": errors, "watchlist_changes": applied_wl}) + + if executed: + try: + updated = await db.get_portfolio() + pv = sum( + p["quantity"] * (state.price_cache.get_price(p["ticker"]) or p["avg_cost"]) + for p in updated["positions"] + ) + await db.record_snapshot(updated["cash"] + pv) + except Exception: + logger.exception("Failed to record post-trade snapshot") + + await db.save_message("assistant", response_message, actions) + + return { + "message": response_message, + "trades": executed, + "trade_errors": errors, + "watchlist_changes": applied_wl, + } diff --git a/backend/app/api/health.py b/backend/app/api/health.py new file mode 100644 index 0000000..8a7894a --- /dev/null +++ b/backend/app/api/health.py @@ -0,0 +1,9 @@ +"""Health check endpoint.""" +from fastapi import APIRouter + +router = APIRouter(prefix="/api", tags=["system"]) + + +@router.get("/health") +async def health() -> dict: + return {"status": "ok"} diff --git a/backend/app/api/portfolio.py b/backend/app/api/portfolio.py new file mode 100644 index 0000000..e931b23 --- /dev/null +++ b/backend/app/api/portfolio.py @@ -0,0 +1,109 @@ +"""Portfolio REST API endpoints.""" +from __future__ import annotations + +import logging +from typing import Annotated + +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel, Field + +from app import db +from app import state + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/portfolio", tags=["portfolio"]) + + +class TradeRequest(BaseModel): + ticker: str + side: str + quantity: float = Field(gt=0) + + +@router.get("") +async def get_portfolio() -> dict: + """Return positions, cash, total value, and unrealized P&L.""" + data = await db.get_portfolio() + cash: float = data["cash"] + positions: list[dict] = data["positions"] + + enriched = [] + positions_value = 0.0 + for pos in positions: + ticker = pos["ticker"] + qty: float = pos["quantity"] + avg_cost: float = pos["avg_cost"] + current = state.price_cache.get_price(ticker) + + if current is not None: + market_value = qty * current + unrealized_pnl = (current - avg_cost) * qty + pnl_percent = (current - avg_cost) / avg_cost * 100 + else: + market_value = qty * avg_cost + unrealized_pnl = 0.0 + pnl_percent = 0.0 + + positions_value += market_value + enriched.append( + { + "ticker": ticker, + "quantity": round(qty, 4), + "avg_cost": round(avg_cost, 4), + "current_price": current, + "market_value": round(market_value, 2), + "unrealized_pnl": round(unrealized_pnl, 2), + "pnl_percent": round(pnl_percent, 2), + "updated_at": pos["updated_at"], + } + ) + + total_value = cash + positions_value + return { + "cash": round(cash, 2), + "positions": enriched, + "positions_value": round(positions_value, 2), + "total_value": round(total_value, 2), + } + + +@router.post("/trade") +async def execute_trade(req: TradeRequest) -> dict: + """Execute a market order at the current price.""" + ticker = req.ticker.upper().strip() + side = req.side.lower() + + if side not in ("buy", "sell"): + raise HTTPException(status_code=422, detail="side must be 'buy' or 'sell'") + + price = state.price_cache.get_price(ticker) + if price is None: + raise HTTPException(status_code=422, detail=f"No price available for {ticker}") + + result = await db.execute_trade(ticker, side, req.quantity, price) + + if "error" in result: + raise HTTPException(status_code=422, detail=result["error"]) + + # Record a portfolio snapshot immediately after the trade + try: + portfolio = await db.get_portfolio() + pv = sum( + pos["quantity"] * (state.price_cache.get_price(pos["ticker"]) or pos["avg_cost"]) + for pos in portfolio["positions"] + ) + await db.record_snapshot(portfolio["cash"] + pv) + except Exception: + logger.exception("Failed to record snapshot after trade") + + return result + + +@router.get("/history") +async def get_history( + limit: Annotated[int, Query(ge=1, le=5000)] = 500, +) -> dict: + """Return portfolio value snapshots over time.""" + snapshots = await db.get_snapshots(limit) + return {"snapshots": snapshots} diff --git a/backend/app/api/watchlist.py b/backend/app/api/watchlist.py new file mode 100644 index 0000000..e40e6e1 --- /dev/null +++ b/backend/app/api/watchlist.py @@ -0,0 +1,65 @@ +"""Watchlist REST API endpoints.""" +from __future__ import annotations + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from app import db +from app import state + +router = APIRouter(prefix="/api/watchlist", tags=["watchlist"]) + + +class AddTickerRequest(BaseModel): + ticker: str + + +@router.get("") +async def get_watchlist() -> dict: + """Return all watchlist tickers with current prices.""" + tickers = await db.get_watchlist() + result = [] + for ticker in tickers: + update = state.price_cache.get(ticker) + result.append( + { + "ticker": ticker, + "price": update.price if update else None, + "change": update.change if update else None, + "change_percent": update.change_percent if update else None, + "direction": update.direction if update else None, + } + ) + return {"tickers": result} + + +@router.post("", status_code=201) +async def add_ticker(req: AddTickerRequest) -> dict: + """Add a ticker to the watchlist and start price tracking.""" + ticker = req.ticker.upper().strip() + if not ticker.isalpha() or len(ticker) > 10: + raise HTTPException(status_code=422, detail="Invalid ticker symbol") + + added = await db.add_ticker(ticker) + if not added: + raise HTTPException(status_code=409, detail=f"{ticker} already in watchlist") + + if state.market_source is not None: + await state.market_source.add_ticker(ticker) + + return {"ticker": ticker, "added": True} + + +@router.delete("/{ticker}", status_code=200) +async def remove_ticker(ticker: str) -> dict: + """Remove a ticker from the watchlist and stop price tracking.""" + ticker = ticker.upper().strip() + removed = await db.remove_ticker(ticker) + if not removed: + raise HTTPException(status_code=404, detail=f"{ticker} not in watchlist") + + if state.market_source is not None: + await state.market_source.remove_ticker(ticker) + state.price_cache.remove(ticker) + + return {"ticker": ticker, "removed": True} diff --git a/backend/app/db.py b/backend/app/db.py new file mode 100644 index 0000000..dae3c90 --- /dev/null +++ b/backend/app/db.py @@ -0,0 +1,354 @@ +"""SQLite database layer for FinAlly. + +All public async functions wrap synchronous SQLite operations via asyncio.to_thread() +to avoid blocking the event loop. init_db_sync() is also exposed for test fixtures. +""" +from __future__ import annotations + +import asyncio +import sqlite3 +import uuid +from datetime import datetime, timezone +from pathlib import Path + +DB_PATH: Path = Path("db/finally.db") + +_DEFAULT_USER = "default" +_DEFAULT_CASH = 10_000.0 +_DEFAULT_TICKERS = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "NVDA", "META", "JPM", "V", "NFLX"] + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS users_profile ( + id TEXT PRIMARY KEY, + cash_balance REAL NOT NULL DEFAULT 10000.0, + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS watchlist ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL DEFAULT 'default', + ticker TEXT NOT NULL, + added_at TEXT NOT NULL, + UNIQUE(user_id, ticker) +); + +CREATE TABLE IF NOT EXISTS positions ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL DEFAULT 'default', + ticker TEXT NOT NULL, + quantity REAL NOT NULL, + avg_cost REAL NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(user_id, ticker) +); + +CREATE TABLE IF NOT EXISTS trades ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL DEFAULT 'default', + ticker TEXT NOT NULL, + side TEXT NOT NULL, + quantity REAL NOT NULL, + price REAL NOT NULL, + executed_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS portfolio_snapshots ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL DEFAULT 'default', + total_value REAL NOT NULL, + recorded_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS chat_messages ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL DEFAULT 'default', + role TEXT NOT NULL, + content TEXT NOT NULL, + actions TEXT, + created_at TEXT NOT NULL +); +""" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _connect(db_path: Path) -> sqlite3.Connection: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + +def init_db_sync(db_path: Path | None = None) -> None: + """Initialize schema and seed default data. Safe to call multiple times.""" + path = db_path or DB_PATH + conn = _connect(path) + try: + with conn: + conn.executescript(_SCHEMA) + conn.execute( + "INSERT OR IGNORE INTO users_profile (id, cash_balance, created_at) VALUES (?, ?, ?)", + (_DEFAULT_USER, _DEFAULT_CASH, _now()), + ) + count = conn.execute( + "SELECT COUNT(*) FROM watchlist WHERE user_id = ?", (_DEFAULT_USER,) + ).fetchone()[0] + if count == 0: + for ticker in _DEFAULT_TICKERS: + conn.execute( + "INSERT OR IGNORE INTO watchlist (id, user_id, ticker, added_at) VALUES (?, ?, ?, ?)", + (str(uuid.uuid4()), _DEFAULT_USER, ticker, _now()), + ) + finally: + conn.close() + + +async def init_db() -> None: + """Async: initialize the database using the current DB_PATH.""" + await asyncio.to_thread(init_db_sync) + + +# ─── Watchlist ───────────────────────────────────────────────────────────────── + +def _get_watchlist_sync() -> list[str]: + conn = _connect(DB_PATH) + try: + rows = conn.execute( + "SELECT ticker FROM watchlist WHERE user_id = ? ORDER BY added_at", + (_DEFAULT_USER,), + ).fetchall() + return [r["ticker"] for r in rows] + finally: + conn.close() + + +async def get_watchlist() -> list[str]: + return await asyncio.to_thread(_get_watchlist_sync) + + +def _add_ticker_sync(ticker: str) -> bool: + conn = _connect(DB_PATH) + try: + with conn: + try: + conn.execute( + "INSERT INTO watchlist (id, user_id, ticker, added_at) VALUES (?, ?, ?, ?)", + (str(uuid.uuid4()), _DEFAULT_USER, ticker, _now()), + ) + return True + except sqlite3.IntegrityError: + return False + finally: + conn.close() + + +async def add_ticker(ticker: str) -> bool: + return await asyncio.to_thread(_add_ticker_sync, ticker) + + +def _remove_ticker_sync(ticker: str) -> bool: + conn = _connect(DB_PATH) + try: + with conn: + cursor = conn.execute( + "DELETE FROM watchlist WHERE user_id = ? AND ticker = ?", + (_DEFAULT_USER, ticker), + ) + return cursor.rowcount > 0 + finally: + conn.close() + + +async def remove_ticker(ticker: str) -> bool: + return await asyncio.to_thread(_remove_ticker_sync, ticker) + + +# ─── Portfolio ───────────────────────────────────────────────────────────────── + +def _get_portfolio_sync() -> dict: + conn = _connect(DB_PATH) + try: + cash_row = conn.execute( + "SELECT cash_balance FROM users_profile WHERE id = ?", (_DEFAULT_USER,) + ).fetchone() + if cash_row is None: + raise RuntimeError("Default user profile not found — database not initialized") + positions = conn.execute( + "SELECT ticker, quantity, avg_cost, updated_at FROM positions WHERE user_id = ?", + (_DEFAULT_USER,), + ).fetchall() + return {"cash": cash_row["cash_balance"], "positions": [dict(r) for r in positions]} + finally: + conn.close() + + +async def get_portfolio() -> dict: + return await asyncio.to_thread(_get_portfolio_sync) + + +def _execute_trade_sync(ticker: str, side: str, quantity: float, price: float) -> dict: + conn = _connect(DB_PATH) + try: + with conn: + cash = conn.execute( + "SELECT cash_balance FROM users_profile WHERE id = ?", (_DEFAULT_USER,) + ).fetchone()["cash_balance"] + + if side == "buy": + cost = quantity * price + if cost > cash: + return {"error": f"Insufficient cash: need ${cost:.2f}, have ${cash:.2f}"} + + conn.execute( + "UPDATE users_profile SET cash_balance = ? WHERE id = ?", + (cash - cost, _DEFAULT_USER), + ) + + existing = conn.execute( + "SELECT quantity, avg_cost FROM positions WHERE user_id = ? AND ticker = ?", + (_DEFAULT_USER, ticker), + ).fetchone() + + if existing: + old_qty = existing["quantity"] + old_avg = existing["avg_cost"] + new_qty = old_qty + quantity + new_avg = (old_qty * old_avg + quantity * price) / new_qty + conn.execute( + "UPDATE positions SET quantity = ?, avg_cost = ?, updated_at = ? " + "WHERE user_id = ? AND ticker = ?", + (new_qty, new_avg, _now(), _DEFAULT_USER, ticker), + ) + else: + conn.execute( + "INSERT INTO positions (id, user_id, ticker, quantity, avg_cost, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (str(uuid.uuid4()), _DEFAULT_USER, ticker, quantity, price, _now()), + ) + + elif side == "sell": + existing = conn.execute( + "SELECT quantity FROM positions WHERE user_id = ? AND ticker = ?", + (_DEFAULT_USER, ticker), + ).fetchone() + if not existing: + return {"error": f"No position in {ticker}"} + held = existing["quantity"] + if held < quantity: + return {"error": f"Insufficient shares: have {held:.4f}, selling {quantity:.4f}"} + + conn.execute( + "UPDATE users_profile SET cash_balance = ? WHERE id = ?", + (cash + quantity * price, _DEFAULT_USER), + ) + new_qty = held - quantity + if new_qty < 1e-9: + conn.execute( + "DELETE FROM positions WHERE user_id = ? AND ticker = ?", + (_DEFAULT_USER, ticker), + ) + else: + conn.execute( + "UPDATE positions SET quantity = ?, updated_at = ? " + "WHERE user_id = ? AND ticker = ?", + (new_qty, _now(), _DEFAULT_USER, ticker), + ) + + else: + return {"error": f"Invalid side: {side}"} + + trade_id = str(uuid.uuid4()) + executed_at = _now() + conn.execute( + "INSERT INTO trades (id, user_id, ticker, side, quantity, price, executed_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (trade_id, _DEFAULT_USER, ticker, side, quantity, price, executed_at), + ) + return { + "trade_id": trade_id, + "ticker": ticker, + "side": side, + "quantity": quantity, + "price": price, + "executed_at": executed_at, + } + finally: + conn.close() + + +async def execute_trade(ticker: str, side: str, quantity: float, price: float) -> dict: + return await asyncio.to_thread(_execute_trade_sync, ticker, side, quantity, price) + + +def _record_snapshot_sync(total_value: float) -> None: + conn = _connect(DB_PATH) + try: + with conn: + conn.execute( + "INSERT INTO portfolio_snapshots (id, user_id, total_value, recorded_at) " + "VALUES (?, ?, ?, ?)", + (str(uuid.uuid4()), _DEFAULT_USER, total_value, _now()), + ) + finally: + conn.close() + + +async def record_snapshot(total_value: float) -> None: + await asyncio.to_thread(_record_snapshot_sync, total_value) + + +def _get_snapshots_sync(limit: int) -> list[dict]: + conn = _connect(DB_PATH) + try: + rows = conn.execute( + "SELECT total_value, recorded_at FROM portfolio_snapshots " + "WHERE user_id = ? ORDER BY recorded_at DESC LIMIT ?", + (_DEFAULT_USER, limit), + ).fetchall() + return [dict(r) for r in reversed(rows)] + finally: + conn.close() + + +async def get_snapshots(limit: int = 500) -> list[dict]: + return await asyncio.to_thread(_get_snapshots_sync, limit) + + +# ─── Chat ────────────────────────────────────────────────────────────────────── + +def _get_chat_history_sync(limit: int) -> list[dict]: + conn = _connect(DB_PATH) + try: + rows = conn.execute( + "SELECT role, content, actions, created_at FROM chat_messages " + "WHERE user_id = ? ORDER BY created_at DESC LIMIT ?", + (_DEFAULT_USER, limit), + ).fetchall() + return [dict(r) for r in reversed(rows)] + finally: + conn.close() + + +async def get_chat_history(limit: int = 20) -> list[dict]: + return await asyncio.to_thread(_get_chat_history_sync, limit) + + +def _save_message_sync(role: str, content: str, actions: str | None) -> None: + conn = _connect(DB_PATH) + try: + with conn: + conn.execute( + "INSERT INTO chat_messages (id, user_id, role, content, actions, created_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (str(uuid.uuid4()), _DEFAULT_USER, role, content, actions, _now()), + ) + finally: + conn.close() + + +async def save_message(role: str, content: str, actions: str | None = None) -> None: + await asyncio.to_thread(_save_message_sync, role, content, actions) diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000..397c5c2 --- /dev/null +++ b/backend/app/main.py @@ -0,0 +1,77 @@ +"""FastAPI application entry point.""" +from __future__ import annotations + +import asyncio +import logging +import os +from contextlib import asynccontextmanager +from pathlib import Path + +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles + +from app import db +from app import state +from app.api import health, portfolio, watchlist, chat +from app.market import create_market_data_source, create_stream_router + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(name)s %(levelname)s %(message)s", +) +logger = logging.getLogger(__name__) + + +async def _snapshot_loop(interval: float = 30.0) -> None: + """Background task: record portfolio value every 30 seconds.""" + while True: + await asyncio.sleep(interval) + try: + data = await db.get_portfolio() + pv = sum( + pos["quantity"] * (state.price_cache.get_price(pos["ticker"]) or pos["avg_cost"]) + for pos in data["positions"] + ) + await db.record_snapshot(data["cash"] + pv) + except Exception: + logger.exception("Snapshot loop error") + + +@asynccontextmanager +async def lifespan(_app: FastAPI): + db_path = Path(os.environ.get("DB_PATH", "db/finally.db")) + db.DB_PATH = db_path + + await db.init_db() + logger.info("Database initialized at %s", db_path) + + tickers = await db.get_watchlist() + state.market_source = create_market_data_source(state.price_cache) + await state.market_source.start(tickers) + logger.info("Market data started for %d tickers", len(tickers)) + + snapshot_task = asyncio.create_task(_snapshot_loop()) + + yield + + snapshot_task.cancel() + try: + await snapshot_task + except asyncio.CancelledError: + pass + + await state.market_source.stop() + logger.info("Shutdown complete") + + +app = FastAPI(title="FinAlly API", version="0.1.0", lifespan=lifespan) + +app.include_router(health.router) +app.include_router(portfolio.router) +app.include_router(watchlist.router) +app.include_router(chat.router) +app.include_router(create_stream_router(state.price_cache)) + +_static = Path("static") +if _static.exists(): + app.mount("/", StaticFiles(directory=str(_static), html=True), name="static") diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b..c6e2d48 100644 --- a/backend/app/market/stream.py +++ b/backend/app/market/stream.py @@ -1,5 +1,4 @@ """SSE streaming endpoint for live price updates.""" - from __future__ import annotations import asyncio @@ -30,7 +29,7 @@ async def stream_prices(request: Request) -> StreamingResponse: Streams all tracked ticker prices every ~500ms. The client connects with EventSource and receives events in the format: - data: {"AAPL": {"ticker": "AAPL", "price": 190.50, ...}, ...} + data: {"AAPL": {"ticker": "AAPL", "price": 190.50, "open_price": 189.00, ...}, ...} Includes a retry directive so the browser auto-reconnects on disconnection (EventSource built-in behavior). @@ -57,11 +56,15 @@ async def _generate_events( Sends all prices every `interval` seconds. Stops when the client disconnects (detected via request.is_disconnected()). + + Tracks open_price per ticker — the first price seen since this SSE + connection was established. Used by the frontend for session change %. """ # Tell the client to retry after 1 second if the connection drops yield "retry: 1000\n\n" last_version = -1 + open_prices: dict[str, float] = {} client_ip = request.client.host if request.client else "unknown" logger.info("SSE client connected: %s", client_ip) @@ -78,9 +81,15 @@ async def _generate_events( prices = price_cache.get_all() if prices: - data = {ticker: update.to_dict() for ticker, update in prices.items()} - payload = json.dumps(data) - yield f"data: {payload}\n\n" + data = {} + for ticker, update in prices.items(): + # First price seen for this ticker on this connection = open price + if ticker not in open_prices: + open_prices[ticker] = update.price + entry = update.to_dict() + entry["open_price"] = open_prices[ticker] + data[ticker] = entry + yield f"data: {json.dumps(data)}\n\n" await asyncio.sleep(interval) except asyncio.CancelledError: diff --git a/backend/app/state.py b/backend/app/state.py new file mode 100644 index 0000000..8809805 --- /dev/null +++ b/backend/app/state.py @@ -0,0 +1,10 @@ +"""Mutable global application state. + +Initialized to defaults; overwritten during the FastAPI lifespan startup. +""" +from __future__ import annotations + +from app.market import PriceCache + +price_cache: PriceCache = PriceCache() +market_source = None # MarketDataSource, set during lifespan startup diff --git a/backend/pyproject.toml b/backend/pyproject.toml index e172cca..8ac9542 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "numpy>=2.0.0", "massive>=1.0.0", "rich>=13.0.0", + "litellm>=1.0.0", ] [project.optional-dependencies] @@ -18,6 +19,7 @@ dev = [ "pytest-asyncio>=0.24.0", "pytest-cov>=5.0.0", "ruff>=0.7.0", + "httpx>=0.27.0", ] [build-system] diff --git a/backend/tests/api/__init__.py b/backend/tests/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/api/conftest.py b/backend/tests/api/conftest.py new file mode 100644 index 0000000..4f2a8f7 --- /dev/null +++ b/backend/tests/api/conftest.py @@ -0,0 +1,82 @@ +"""Shared fixtures for API endpoint tests.""" +from __future__ import annotations + +from contextlib import asynccontextmanager +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +import app.db as db_module +import app.state as state +from app.market import PriceCache + +_TICKERS = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "NVDA", "META", "JPM", "V", "NFLX"] +_PRICES = {t: 100.0 + i * 10 for i, t in enumerate(_TICKERS)} + + +class _MockMarketSource: + def __init__(self) -> None: + self.added: list[str] = [] + self.removed: list[str] = [] + + async def add_ticker(self, ticker: str) -> None: + self.added.append(ticker) + + async def remove_ticker(self, ticker: str) -> None: + self.removed.append(ticker) + + async def start(self, tickers: list[str]) -> None: + pass + + async def stop(self) -> None: + pass + + +@pytest.fixture +def mock_market_source() -> _MockMarketSource: + return _MockMarketSource() + + +@pytest.fixture +def mock_price_cache() -> PriceCache: + cache = PriceCache() + for ticker, price in _PRICES.items(): + cache.update(ticker, price) + return cache + + +@pytest.fixture +def tmp_db(tmp_path: Path) -> Path: + """Temporary SQLite database initialized with schema and seed data.""" + db_path = tmp_path / "test.db" + original = db_module.DB_PATH + db_module.DB_PATH = db_path + db_module.init_db_sync(db_path) + yield db_path + db_module.DB_PATH = original + + +@pytest.fixture +def client( + tmp_db: Path, + mock_price_cache: PriceCache, + mock_market_source: _MockMarketSource, +): + """TestClient with mocked lifespan injecting test state.""" + from app.main import app as fastapi_app + + original_lifespan = fastapi_app.router.lifespan_context + + @asynccontextmanager + async def _mock_lifespan(_app): + state.price_cache = mock_price_cache + state.market_source = mock_market_source + yield + + fastapi_app.router.lifespan_context = _mock_lifespan + + with TestClient(fastapi_app) as c: + yield c + + fastapi_app.router.lifespan_context = original_lifespan diff --git a/backend/tests/api/test_health.py b/backend/tests/api/test_health.py new file mode 100644 index 0000000..4649f6c --- /dev/null +++ b/backend/tests/api/test_health.py @@ -0,0 +1,7 @@ +"""Tests for GET /api/health.""" + + +def test_health_returns_ok(client): + resp = client.get("/api/health") + assert resp.status_code == 200 + assert resp.json() == {"status": "ok"} diff --git a/backend/tests/api/test_portfolio.py b/backend/tests/api/test_portfolio.py new file mode 100644 index 0000000..11e0946 --- /dev/null +++ b/backend/tests/api/test_portfolio.py @@ -0,0 +1,108 @@ +"""Tests for portfolio API endpoints.""" +from __future__ import annotations + + +def test_get_portfolio_initial_state(client): + resp = client.get("/api/portfolio") + assert resp.status_code == 200 + data = resp.json() + assert data["cash"] == 10_000.0 + assert data["positions"] == [] + assert data["positions_value"] == 0.0 + assert data["total_value"] == 10_000.0 + + +def test_get_portfolio_history_empty(client): + resp = client.get("/api/portfolio/history") + assert resp.status_code == 200 + assert resp.json()["snapshots"] == [] + + +def test_buy_creates_position(client): + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 5} + ) + assert resp.status_code == 200 + data = resp.json() + assert data["ticker"] == "AAPL" + assert data["side"] == "buy" + assert data["quantity"] == 5 + assert "trade_id" in data + + portfolio = client.get("/api/portfolio").json() + assert portfolio["cash"] < 10_000.0 + assert len(portfolio["positions"]) == 1 + assert portfolio["positions"][0]["ticker"] == "AAPL" + assert portfolio["positions"][0]["quantity"] == 5 + + +def test_sell_without_position_returns_422(client): + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "sell", "quantity": 1} + ) + assert resp.status_code == 422 + + +def test_buy_insufficient_cash_returns_422(client): + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 1_000_000} + ) + assert resp.status_code == 422 + + +def test_invalid_side_returns_422(client): + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "hold", "quantity": 1} + ) + assert resp.status_code == 422 + + +def test_unknown_ticker_returns_422(client): + resp = client.post( + "/api/portfolio/trade", json={"ticker": "ZZZZ", "side": "buy", "quantity": 1} + ) + assert resp.status_code == 422 + + +def test_buy_then_full_sell_removes_position(client): + client.post("/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 10}) + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "sell", "quantity": 10} + ) + assert resp.status_code == 200 + + portfolio = client.get("/api/portfolio").json() + assert portfolio["positions"] == [] + # Cash should be back to ~10000 (buy and sell at same simulated price) + assert abs(portfolio["cash"] - 10_000.0) < 0.01 + + +def test_sell_more_than_held_returns_422(client): + client.post("/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 5}) + resp = client.post( + "/api/portfolio/trade", json={"ticker": "AAPL", "side": "sell", "quantity": 100} + ) + assert resp.status_code == 422 + + +def test_portfolio_snapshot_recorded_after_trade(client): + client.post("/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 1}) + resp = client.get("/api/portfolio/history") + assert resp.status_code == 200 + assert len(resp.json()["snapshots"]) >= 1 + + +def test_portfolio_history_limit_param(client): + resp = client.get("/api/portfolio/history?limit=5") + assert resp.status_code == 200 + assert isinstance(resp.json()["snapshots"], list) + + +def test_buy_updates_avg_cost_on_second_purchase(client): + client.post("/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 10}) + client.post("/api/portfolio/trade", json={"ticker": "AAPL", "side": "buy", "quantity": 10}) + portfolio = client.get("/api/portfolio").json() + pos = portfolio["positions"][0] + assert pos["quantity"] == 20 + # avg_cost should equal the simulated price (price doesn't change in mock) + assert pos["avg_cost"] == pos["current_price"] diff --git a/backend/tests/api/test_watchlist.py b/backend/tests/api/test_watchlist.py new file mode 100644 index 0000000..b94864e --- /dev/null +++ b/backend/tests/api/test_watchlist.py @@ -0,0 +1,80 @@ +"""Tests for watchlist API endpoints.""" +from __future__ import annotations + + +def test_get_watchlist_returns_default_tickers(client): + resp = client.get("/api/watchlist") + assert resp.status_code == 200 + data = resp.json() + tickers = [t["ticker"] for t in data["tickers"]] + assert "AAPL" in tickers + assert "NVDA" in tickers + assert len(tickers) == 10 + + +def test_get_watchlist_includes_prices(client): + resp = client.get("/api/watchlist") + for item in resp.json()["tickers"]: + # All default tickers are in the mock cache + assert item["price"] is not None + assert item["direction"] in ("up", "down", "flat") + + +def test_add_ticker(client, mock_market_source): + resp = client.post("/api/watchlist", json={"ticker": "PYPL"}) + assert resp.status_code == 201 + assert resp.json()["ticker"] == "PYPL" + assert resp.json()["added"] is True + assert "PYPL" in mock_market_source.added + + tickers = [t["ticker"] for t in client.get("/api/watchlist").json()["tickers"]] + assert "PYPL" in tickers + + +def test_add_ticker_lowercase_is_normalized(client): + # 'pypl' normalized to 'PYPL' which isn't in default watchlist, so should succeed + resp = client.post("/api/watchlist", json={"ticker": "pypl"}) + assert resp.status_code == 201 + assert resp.json()["ticker"] == "PYPL" + + +def test_add_duplicate_ticker_returns_409(client): + resp = client.post("/api/watchlist", json={"ticker": "AAPL"}) + assert resp.status_code == 409 + + +def test_add_invalid_ticker_returns_422(client): + for bad in ["AAPL123", "!!", "TOOLONGTICKER", ""]: + resp = client.post("/api/watchlist", json={"ticker": bad}) + assert resp.status_code == 422, f"Expected 422 for ticker '{bad}'" + + +def test_remove_ticker(client, mock_market_source): + resp = client.delete("/api/watchlist/AAPL") + assert resp.status_code == 200 + assert resp.json()["removed"] is True + assert "AAPL" in mock_market_source.removed + + tickers = [t["ticker"] for t in client.get("/api/watchlist").json()["tickers"]] + assert "AAPL" not in tickers + + +def test_remove_nonexistent_ticker_returns_404(client): + resp = client.delete("/api/watchlist/FAKE") + assert resp.status_code == 404 + + +def test_remove_ticker_lowercase_is_normalized(client): + resp = client.delete("/api/watchlist/aapl") + assert resp.status_code == 200 + assert resp.json()["ticker"] == "AAPL" + + +def test_add_then_remove_roundtrip(client): + client.post("/api/watchlist", json={"ticker": "HOOD"}) + tickers_after_add = [t["ticker"] for t in client.get("/api/watchlist").json()["tickers"]] + assert "HOOD" in tickers_after_add + + client.delete("/api/watchlist/HOOD") + tickers_after_remove = [t["ticker"] for t in client.get("/api/watchlist").json()["tickers"]] + assert "HOOD" not in tickers_after_remove