Add finance-cluster GraphQL integration

README.md

@@ -23,7 +23,11 @@ src/
   server.js        # Apollo Server factory (reused by the tests)
   schema.js        # GraphQL type definitions
   resolvers.js     # Query / Mutation / field resolvers
+  config/finance.js # Environment-driven finance connector config
+  connectors/      # Replaceable OpenTrading, Portfolio-Watcher, tax-break adapters
   data/store.js    # In-memory data store with seed data
+  domain/finance.js # Canonical finance models and normalization helpers
+  services/financeService.js # Finance aggregation, caching, and error handling
 test/
   graphql.test.js  # API tests executed against the schema
 website/
@@ -99,6 +103,108 @@ npm run lint                      # Oxlint
 npm test
 ```
 
+## Finance Cluster Integration (Priority 1)
+
+This service now exposes a unified finance GraphQL surface over three upstream
+domains:
+
+- **OpenTrading**: accounts, orders, trades, and fills
+- **Portfolio-Watcher**: holdings/positions and performance snapshots
+- **tax-break**: trade-to-tax-event mapping and tax estimate summaries
+
+The initial implementation uses mock connectors under `src/connectors/` so the
+API runs from a clean checkout. Each connector exposes a small async contract
+that can be replaced later with HTTP, gRPC, queue, or database-backed clients
+without changing the GraphQL schema.
+
+### Configuration
+
+The running server loads connector settings from the environment via
+`src/config/finance.js`:
+
+| Variable | Description | Default |
+| --- | --- | --- |
+| `OPENTRADING_ENDPOINT` | OpenTrading endpoint placeholder | `mock://opentrading` |
+| `OPENTRADING_API_KEY` | OpenTrading credential placeholder | empty |
+| `PORTFOLIO_WATCHER_ENDPOINT` | Portfolio-Watcher endpoint placeholder | `mock://portfolio-watcher` |
+| `PORTFOLIO_WATCHER_API_KEY` | Portfolio-Watcher credential placeholder | empty |
+| `TAX_BREAK_ENDPOINT` | tax-break endpoint placeholder | `mock://tax-break` |
+| `TAX_BREAK_API_KEY` | tax-break credential placeholder | empty |
+| `FINANCE_CACHE_TTL_MS` | Minimal resolver cache TTL | `1000` |
+
+Do not commit real credentials. Production connectors should read credentials
+from environment variables or a secret manager and keep the same method names as
+the mock adapters.
+
+### Data flow
+
+1. GraphQL resolvers call `financeService` through the request context.
+2. `financeService` calls each upstream connector and normalizes inconsistent
+   field names in `src/domain/finance.js`.
+3. OpenTrading trades are enriched through tax-break into `TaxEvent` records.
+4. Portfolio-Watcher positions and snapshots are aggregated with accounts into a
+   portfolio overview with total market value and unrealized P/L.
+5. Connector failures are captured as `FinanceUpstreamError` objects so clients
+   receive actionable source/code/message details while still getting any
+   partial data from healthy upstreams.
+
+### Finance queries
+
+Portfolio overview with positions and P/L:
+
+```graphql
+query PortfolioOverview {
+  portfolioOverview {
+    accounts { id name provider currency }
+    positions { symbol quantity marketValue unrealizedPnL }
+    performance { asOf totalValue dayPnL totalPnL }
+    totalMarketValue
+    totalUnrealizedPnL
+    errors { source code message }
+  }
+}
+```
+
+Trade history mapped to tax-relevant events:
+
+```graphql
+query TradeHistory {
+  tradeHistory(symbol: "AAPL") {
+    trades { id side symbol quantity price executedAt }
+    taxEvents { tradeId proceeds costBasis realizedGain occurredAt }
+    errors { source code message }
+  }
+}
+```
+
+Tax summary traceable to the underlying trading activity:
+
+```graphql
+query TaxEstimate {
+  taxEstimate(taxYear: 2026) {
+    totalProceeds
+    totalCostBasis
+    realizedGain
+    estimatedTax
+    events { id tradeId realizedGain }
+    errors { source code message }
+  }
+}
+```
+
+Run the API and tests with the existing commands:
+
+```bash
+npm start
+npm test
+```
+
+Follow-up production tasks:
+
+- Replace mock connectors with authenticated clients for each upstream domain.
+- Add pagination/date filters once live trade and snapshot volumes grow.
+- Add persisted caching/batching if upstream latency becomes significant.
+
 ## API
 
 | Operation | Description |
@@ -107,6 +213,9 @@ npm test
 | `user(id: ID!)` | Fetch a single user, `null` when unknown |
 | `posts` | List all posts |
 | `post(id: ID!)` | Fetch a single post, `null` when unknown |
+| `portfolioOverview(accountId)` | Fetch finance accounts, positions, snapshots, and P/L |
+| `tradeHistory(accountId, symbol)` | Fetch trades/orders enriched with tax events |
+| `taxEstimate(taxYear, accountId)` | Estimate tax from tax-relevant trading activity |
 | `createUser(name, email)` | Create a user |
 | `createPost(title, content, authorId)` | Create a post for an existing user |
 

src/config/finance.js

@@ -0,0 +1,18 @@
+/** Environment-driven configuration for finance-cluster connectors. */
+export function loadFinanceConfig(env = process.env) {
+  return {
+    openTrading: {
+      endpoint: env.OPENTRADING_ENDPOINT ?? 'mock://opentrading',
+      apiKey: env.OPENTRADING_API_KEY ?? '',
+    },
+    portfolioWatcher: {
+      endpoint: env.PORTFOLIO_WATCHER_ENDPOINT ?? 'mock://portfolio-watcher',
+      apiKey: env.PORTFOLIO_WATCHER_API_KEY ?? '',
+    },
+    taxBreak: {
+      endpoint: env.TAX_BREAK_ENDPOINT ?? 'mock://tax-break',
+      apiKey: env.TAX_BREAK_API_KEY ?? '',
+    },
+    cacheTtlMs: Number(env.FINANCE_CACHE_TTL_MS ?? 1000),
+  };
+}

src/connectors/opentrading/mockConnector.js

@@ -0,0 +1,45 @@
+/** Mock OpenTrading connector. Replace this contract with HTTP/gRPC calls later. */
+export function createOpenTradingConnector(_config = {}) {
+  const accounts = [
+    { acct_id: 'acct-1', display_name: 'Primary Brokerage', account_type: 'BROKERAGE', base_currency: 'USD', source: 'OpenTrading' },
+  ];
+
+  const trades = [
+    {
+      trade_id: 'trade-1',
+      acct_id: 'acct-1',
+      order_ref: 'order-1',
+      ticker: 'AAPL',
+      action: 'BUY',
+      qty: 10,
+      avg_px: 150,
+      trade_time: '2026-01-10T14:31:00.000Z',
+      status: 'FILLED',
+      fills: [{ fill_id: 'fill-1', fill_qty: 10, fill_px: 150, filled_at: '2026-01-10T14:31:00.000Z' }],
+    },
+    {
+      trade_id: 'trade-2',
+      acct_id: 'acct-1',
+      order_ref: 'order-2',
+      ticker: 'AAPL',
+      action: 'SELL',
+      qty: 4,
+      avg_px: 180,
+      trade_time: '2026-04-05T15:45:00.000Z',
+      status: 'FILLED',
+      fills: [{ fill_id: 'fill-2', fill_qty: 4, fill_px: 180, filled_at: '2026-04-05T15:45:00.000Z' }],
+    },
+  ];
+
+  const orders = [
+    { order_id: 'order-1', acct_id: 'acct-1', ticker: 'AAPL', action: 'BUY', qty: 10, limit_price: 150, status: 'FILLED', created_at: '2026-01-10T14:30:00.000Z', fills: trades[0].fills },
+    { order_id: 'order-2', acct_id: 'acct-1', ticker: 'AAPL', action: 'SELL', qty: 4, limit_price: 180, status: 'FILLED', created_at: '2026-04-05T15:40:00.000Z', fills: trades[1].fills },
+  ];
+
+  return {
+    source: 'OpenTrading',
+    listAccounts: async () => accounts,
+    listTrades: async () => trades,
+    listOrders: async () => orders,
+  };
+}

src/connectors/portfolio-watcher/mockConnector.js

@@ -0,0 +1,17 @@
+/** Mock Portfolio-Watcher connector for positions and performance snapshots. */
+export function createPortfolioWatcherConnector(_config = {}) {
+  const positions = [
+    { id: 'pos-1', account_id: 'acct-1', ticker: 'AAPL', shares: 6, avg_cost: 150, last_price: 182.5 },
+    { id: 'pos-2', account_id: 'acct-1', ticker: 'MSFT', shares: 3, avg_cost: 320, last_price: 350 },
+  ];
+
+  const snapshots = [
+    { id: 'snap-1', account_id: 'acct-1', as_of: '2026-04-05T21:00:00.000Z', total_value: 2745, cash_balance: 600, market_value: 2145, day_pnl: 32.5, total_pnl: 285 },
+  ];
+
+  return {
+    source: 'Portfolio-Watcher',
+    listPositions: async () => positions,
+    listPerformanceSnapshots: async () => snapshots,
+  };
+}

src/connectors/tax-break/mockConnector.js

@@ -0,0 +1,10 @@
+import { estimateTaxFromEvents, tradeToTaxEvent } from '../../domain/finance.js';
+
+/** Mock tax-break connector for tax-event enrichment and estimate calculations. */
+export function createTaxBreakConnector(_config = {}) {
+  return {
+    source: 'tax-break',
+    mapTradesToTaxEvents: async (trades) => trades.filter((trade) => trade.side === 'SELL').map(tradeToTaxEvent),
+    estimateTax: async ({ events, taxYear }) => estimateTaxFromEvents(events, taxYear, 0.22),
+  };
+}

src/domain/finance.js

@@ -0,0 +1,141 @@
+const money = (value) => Number(Number(value ?? 0).toFixed(2));
+
+export function normalizeAccount(account) {
+  return {
+    id: String(account.id ?? account.account_id ?? account.acct_id),
+    name: account.name ?? account.display_name ?? 'Brokerage Account',
+    type: account.type ?? account.account_type ?? 'BROKERAGE',
+    currency: account.currency ?? account.base_currency ?? 'USD',
+    provider: account.provider ?? account.source ?? 'unknown',
+  };
+}
+
+export function normalizePosition(position) {
+  const quantity = Number(position.quantity ?? position.qty ?? position.shares ?? 0);
+  const averageCost = Number(position.averageCost ?? position.avg_cost ?? position.cost_basis_per_share ?? 0);
+  const marketPrice = Number(position.marketPrice ?? position.last_price ?? position.market_price ?? 0);
+  const marketValue = money(position.marketValue ?? position.market_value ?? quantity * marketPrice);
+  const costBasis = money(quantity * averageCost);
+
+  return {
+    id: String(position.id ?? `${position.accountId ?? position.account_id ?? position.acct_id}:${position.symbol ?? position.ticker}`),
+    accountId: String(position.accountId ?? position.account_id ?? position.acct_id),
+    symbol: String(position.symbol ?? position.ticker).toUpperCase(),
+    quantity,
+    averageCost,
+    marketPrice,
+    marketValue,
+    unrealizedPnL: money(position.unrealizedPnL ?? position.unrealized_pnl ?? marketValue - costBasis),
+  };
+}
+
+export function normalizeFill(fill) {
+  return {
+    id: String(fill.id ?? fill.fill_id),
+    quantity: Number(fill.quantity ?? fill.fill_qty ?? 0),
+    price: Number(fill.price ?? fill.fill_px ?? 0),
+    executedAt: fill.executedAt ?? fill.filled_at,
+  };
+}
+
+export function normalizeTrade(trade) {
+  return {
+    id: String(trade.id ?? trade.trade_id),
+    accountId: String(trade.accountId ?? trade.account_id ?? trade.acct_id),
+    orderId: String(trade.orderId ?? trade.order_ref ?? trade.order_id),
+    symbol: String(trade.symbol ?? trade.ticker).toUpperCase(),
+    side: String(trade.side ?? trade.action).toUpperCase(),
+    quantity: Number(trade.quantity ?? trade.qty ?? 0),
+    price: Number(trade.price ?? trade.avg_px ?? 0),
+    status: trade.status ?? 'FILLED',
+    executedAt: trade.executedAt ?? trade.trade_time,
+    fills: (trade.fills ?? []).map(normalizeFill),
+  };
+}
+
+export function normalizeOrder(order) {
+  return {
+    id: String(order.id ?? order.order_id),
+    accountId: String(order.accountId ?? order.account_id ?? order.acct_id),
+    symbol: String(order.symbol ?? order.ticker).toUpperCase(),
+    side: String(order.side ?? order.action).toUpperCase(),
+    quantity: Number(order.quantity ?? order.qty ?? 0),
+    limitPrice: order.limitPrice ?? order.limit_price ?? null,
+    status: order.status ?? 'UNKNOWN',
+    createdAt: order.createdAt ?? order.created_at,
+    fills: (order.fills ?? []).map(normalizeFill),
+  };
+}
+
+export function normalizePerformanceSnapshot(snapshot) {
+  return {
+    id: String(snapshot.id ?? `${snapshot.accountId ?? snapshot.account_id}:${snapshot.asOf ?? snapshot.as_of}`),
+    accountId: String(snapshot.accountId ?? snapshot.account_id),
+    asOf: snapshot.asOf ?? snapshot.as_of,
+    totalValue: money(snapshot.totalValue ?? snapshot.total_value ?? 0),
+    cash: money(snapshot.cash ?? snapshot.cash_balance ?? 0),
+    marketValue: money(snapshot.marketValue ?? snapshot.market_value ?? 0),
+    dayPnL: money(snapshot.dayPnL ?? snapshot.day_pnl ?? 0),
+    totalPnL: money(snapshot.totalPnL ?? snapshot.total_pnl ?? 0),
+  };
+}
+
+export function tradeToTaxEvent(trade) {
+  const proceeds = trade.side === 'SELL' ? money(trade.quantity * trade.price) : 0;
+  const costBasis = trade.side === 'SELL' ? money(trade.quantity * trade.price * 0.82) : money(trade.quantity * trade.price);
+
+  return {
+    id: `tax-${trade.id}`,
+    tradeId: trade.id,
+    symbol: trade.symbol,
+    quantity: trade.quantity,
+    proceeds,
+    costBasis,
+    realizedGain: money(proceeds - costBasis),
+    holdingPeriod: 'SHORT_TERM',
+    occurredAt: trade.executedAt,
+  };
+}
+
+export function aggregatePortfolio({ accounts = [], positions = [], snapshots = [], errors = [] }) {
+  return {
+    accounts,
+    positions,
+    performance: snapshots,
+    currency: accounts[0]?.currency ?? 'USD',
+    totalMarketValue: money(positions.reduce((sum, position) => sum + position.marketValue, 0)),
+    totalUnrealizedPnL: money(positions.reduce((sum, position) => sum + position.unrealizedPnL, 0)),
+    errors,
+  };
+}
+
+export function filterByAccount(records, accountId) {
+  return accountId ? records.filter((record) => record.accountId === accountId || record.id === accountId) : records;
+}
+
+export function filterTrades(trades, { accountId, symbol } = {}) {
+  return trades.filter((trade) => {
+    const accountMatches = accountId ? trade.accountId === accountId : true;
+    const symbolMatches = symbol ? trade.symbol === symbol.toUpperCase() : true;
+    return accountMatches && symbolMatches;
+  });
+}
+
+export function estimateTaxFromEvents(events, taxYear, rate = 0.22) {
+  const taxableEvents = events.filter((event) => new Date(event.occurredAt).getUTCFullYear() === taxYear);
+  const totalProceeds = money(taxableEvents.reduce((sum, event) => sum + event.proceeds, 0));
+  const totalCostBasis = money(taxableEvents.reduce((sum, event) => sum + event.costBasis, 0));
+  const realizedGain = money(taxableEvents.reduce((sum, event) => sum + event.realizedGain, 0));
+
+  return {
+    taxYear,
+    currency: 'USD',
+    totalProceeds,
+    totalCostBasis,
+    realizedGain,
+    estimatedTax: money(Math.max(realizedGain, 0) * rate),
+    taxRate: rate,
+    events: taxableEvents,
+    errors: [],
+  };
+}

src/index.js

@@ -3,6 +3,7 @@ import cors from 'cors';
 import express from 'express';
 
 import { store } from './data/store.js';
+import { financeService } from './services/financeService.js';
 import { createApolloServer } from './server.js';
 
 const PORT = Number(process.env.PORT) || 4000;
@@ -23,7 +24,7 @@ async function main() {
     express.json(),
     expressMiddleware(apolloServer, {
       // Every request shares the same in-memory store.
-      context: async () => ({ store }),
+      context: async () => ({ store, finance: financeService }),
     })
   );
 

src/resolvers.js

@@ -10,6 +10,9 @@ export const resolvers = {
     user: (_parent, { id }, { store }) => store.getUser(id),
     posts: (_parent, _args, { store }) => store.listPosts(),
     post: (_parent, { id }, { store }) => store.getPost(id),
+    portfolioOverview: (_parent, args, { finance }) => finance.portfolioOverview(args),
+    tradeHistory: (_parent, args, { finance }) => finance.tradeHistory(args),
+    taxEstimate: (_parent, args, { finance }) => finance.taxEstimate(args),
   },
 
   Mutation: {