Why Local-First Trading?
Cloud-based trading platforms add latency, require subscriptions, and expose your strategies to third parties. Our StockMarket Agent runs entirely on your machine.
Architecture
┌─────────────┐ ┌──────────────┐ ┌──────────────┐
│ Data Feed │───>│ Strategy │───>│ Execution │
│ (yfinance) │ │ Engine │ │ Engine │
└─────────────┘ └──────────────┘ └──────────────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌──────────────┐ ┌──────────────┐
│ DuckDB │ │ Risk Guard │ │ Broker API │
│ (Storage) │ │ (Limits) │ │ (Zerodha) │
└─────────────┘ └──────────────┘ └──────────────┘DuckDB for Financial Data
We chose DuckDB over PostgreSQL for several reasons:
python
import duckdb
class MarketDataStore:
def __init__(self, db_path: str = "market_data.duckdb"):
self.conn = duckdb.connect(db_path)
self._init_schema()
def _init_schema(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS ohlcv (
symbol VARCHAR,
timestamp TIMESTAMP,
open DOUBLE,
high DOUBLE,
low DOUBLE,
close DOUBLE,
volume BIGINT,
PRIMARY KEY (symbol, timestamp)
)
""")
def query_candles(
self, symbol: str, start: datetime, end: datetime
) -> pa.Table:
return self.conn.execute("""
SELECT * FROM ohlcv
WHERE symbol = ? AND timestamp BETWEEN ? AND ?
ORDER BY timestamp
""", [symbol, start, end]).arrow()Walk-Forward Backtesting
Our backtesting engine uses walk-forward optimization to prevent overfitting:
python
class WalkForwardBacktester:
def __init__(
self,
strategy: Strategy,
train_window: int = 252, # 1 year
test_window: int = 63, # 3 months
step_size: int = 21 # 1 month
):
self.strategy = strategy
self.train_window = train_window
self.test_window = test_window
self.step_size = step_size
def run(self, data: pd.DataFrame) -> BacktestResult:
results = []
for i in range(0, len(data) - self.train_window - self.test_window, self.step_size):
train_data = data.iloc[i:i + self.train_window]
test_data = data.iloc[i + self.train_window:i + self.train_window + self.test_window]
# Optimize on training data
self.strategy.optimize(train_data)
# Test on out-of-sample data
period_result = self.strategy.backtest(test_data)
results.append(period_result)
return BacktestResult.aggregate(results)Risk Management
Every trade passes through the Risk Guard: