This is Part 2 of a 3-part series. Part 1 is the story of what went wrong and what I learned building this — worth reading first for context. This post is the step-by-step build: Linux setup and the complete working code. Part 3 covers adding a live performance dashboard served over Apache.
This is a complete walkthrough for building a paper-trading bot in Python using the Alpaca API, running on a Linux server. By the end you’ll have a bot that pulls market data, generates buy/sell signals from a trend-following strategy, manages risk with a volatility-scaled stop-loss, executes paper trades automatically, and logs everything for review.
Everything below is real, working code — this is the actual project I run in production (well, paper-production) on my own Ubuntu server.
What you’ll end up with
- A modular Python project: data feed, strategy, risk management, order execution, and logging as separate files
- A backtesting script to test the strategy against historical data before ever running it live
- A live trading loop that polls the market and trades automatically
- Daily performance logging (equity curve + trade history)
Prerequisites
- A Linux server or machine (Ubuntu used throughout this guide, but any distro with Python 3.10+ works)
- Python 3.10 or newer
- A free Alpaca account (no credit card needed for paper trading)
- Basic command-line familiarity
Step 1: Get free Alpaca paper trading API keys
- Sign up at alpaca.markets — it’s free.
- In the Alpaca dashboard, switch to Paper Trading mode (there’s a toggle near the top).
- Generate an API key and secret from the paper trading section. Keep these handy — you’ll need them shortly.
Paper trading uses simulated money against real market data, so you can test everything safely before ever considering real funds.
Step 2: Set up the project directory and virtual environment
mkdir alpaca_paper_bot
cd alpaca_paper_bot
python3 -m venv venv
source venv/bin/activate
That last command activates the virtual environment — you should see (venv) appear at the start of your terminal prompt. You’ll need to run source venv/bin/activate again every time you open a new terminal session to work on this project; it only applies to the current shell.
Step 3: Install dependencies
Create a requirements.txt file:
alpaca-py>=0.33.0
pandas>=2.0.0
python-dotenv>=1.0.0
matplotlib>=3.7.0
Then install:
pip install -r requirements.txt
Step 4: Store your API keys securely
Create a .env file in the project root:
ALPACA_API_KEY=your_key_here
ALPACA_SECRET_KEY=your_secret_here
Replace the placeholders with the keys from Step 1. Never commit this file to version control — if you’re using git, add .env to your .gitignore.
Step 5: Build the project, file by file
Here’s the full project structure we’re building:
alpaca_paper_bot/
├── config.py # settings + API key loading
├── data_feed.py # pulls price data from Alpaca
├── strategy.py # generates BUY/SELL/HOLD signals
├── risk.py # position sizing + stop-loss logic
├── executor.py # submits orders to Alpaca
├── trade_logger.py # logs trades + equity over time
├── backtest.py # tests the strategy on historical data
├── main.py # the live trading loop
├── requirements.txt
└── .env
config.py — central settings
Everything the rest of the project needs is configured here: API keys, strategy parameters, and risk settings.
"""
Central config. Loads API credentials from a .env file so keys never
get hardcoded or committed to source control.
"""
import os
from dotenv import load_dotenv
load_dotenv()
ALPACA_API_KEY = os.getenv("ALPACA_API_KEY", "")
ALPACA_SECRET_KEY = os.getenv("ALPACA_SECRET_KEY", "")
# Always True for this starter project. Only flip this once you deeply
# understand the strategy's behavior and risk — and even then, start small.
PAPER_TRADING = True
# Strategy defaults (override via CLI args in main.py / backtest.py)
SHORT_WINDOW = 20 # short moving average period (in bars)
LONG_WINDOW = 50 # long moving average period (in bars)
ADX_PERIOD = 14 # lookback for trend-strength calculation
ADX_THRESHOLD = 25 # only trade crossovers when ADX is at/above this (trending market)
# Risk defaults
# NOTE: if you run more than one symbol (e.g. ACN and SPY) against the same
# Alpaca account, MAX_POSITION_PCT is evaluated independently by each
# running instance against the SAME shared equity — it does not know about
# the other bot's positions. Two instances both using --risk-sizing could
# each try to commit 10% independently (20% combined intent), and whichever
# one has less available cash at execution time may get a partial fill or
# rejected order from Alpaca. Using a fixed --qty per instance (rather than
# --risk-sizing) sidesteps this entirely and is the simpler default when
# running multiple symbols. If you do want --risk-sizing on multiple
# symbols at once, consider lowering this (e.g. to 0.05) so the combined
# intent across instances stays within a sane total.
MAX_POSITION_PCT = 0.10 # never risk more than 10% of equity on one symbol
# Stop-loss is ATR-based (scales with each symbol's own volatility) rather
# than a flat percentage — a flat % stop is too tight for volatile symbols
# (gets hit by normal noise) and too loose for calm ones. The stop distance
# is entry_price - (ATR_STOP_MULTIPLIER * ATR at time of entry).
# 2.0-3.0 is a common range; higher = wider stop = fewer, later exits.
ATR_STOP_MULTIPLIER = 2.5
if not ALPACA_API_KEY or not ALPACA_SECRET_KEY:
print(
"[config] WARNING: ALPACA_API_KEY / ALPACA_SECRET_KEY not set. "
"Copy .env.example to .env and fill in your paper trading keys."
)
A few of the defaults worth understanding before you change them:
SHORT_WINDOW/LONG_WINDOW— the two moving averages the strategy compares. 20/50 is a common trend-following pairing.ADX_THRESHOLD— ADX (Average Directional Index) measures trend strength. Below 25 is generally considered a weak or choppy market; the strategy only trades when ADX confirms an actual trend.ATR_STOP_MULTIPLIER— the stop-loss distance is based on each symbol’s own volatility (ATR) rather than a flat percentage. More on why below.
data_feed.py — pulling market data
This wraps Alpaca’s data API so the rest of the code doesn’t need to know the API details.
"""
Wraps Alpaca's historical + latest bar data so the rest of the bot doesn't
need to know about the API details.
"""
from datetime import datetime, timedelta
import pandas as pd
from alpaca.data.historical import StockHistoricalDataClient
from alpaca.data.requests import StockBarsRequest, StockLatestBarRequest
from alpaca.data.timeframe import TimeFrame
import config
class DataFeed:
def __init__(self):
self.client = StockHistoricalDataClient(
config.ALPACA_API_KEY, config.ALPACA_SECRET_KEY
)
def get_historical_bars(self, symbol: str, days: int = 180) -> pd.DataFrame:
"""Return a DataFrame of daily bars for the last `days` days."""
request = StockBarsRequest(
symbol_or_symbols=symbol,
timeframe=TimeFrame.Day,
start=datetime.now() - timedelta(days=days),
)
bars = self.client.get_stock_bars(request)
df = bars.df
if df.empty:
raise ValueError(f"No historical data returned for {symbol}")
# df is multi-indexed by (symbol, timestamp) when multiple symbols
# are requested; normalize to a flat index for a single symbol.
if isinstance(df.index, pd.MultiIndex):
df = df.xs(symbol, level=0)
return df
def get_latest_bar(self, symbol: str):
"""Return the most recent bar for a symbol (used in the live loop)."""
request = StockLatestBarRequest(symbol_or_symbols=symbol)
latest = self.client.get_stock_latest_bar(request)
return latest[symbol]
strategy.py — the actual trading logic
This is the heart of the bot: turning price data into a BUY, SELL, or HOLD decision.
"""
Trend-following strategy: moving-average crossover, gated by ADX trend
strength so signals only fire when there's an actual trend to ride (helps
avoid whipsaws in choppy/sideways markets).
Signal logic is STATE-based, not event-based, and ADX gates BOTH entry
and exit:
- BUY whenever SMA20 is above SMA50 AND ADX >= threshold (checked every
bar, not just the bar the crossover happened on)
- SELL whenever SMA20 is below SMA50 AND ADX >= threshold
- HOLD otherwise — including while SMA has crossed below but ADX hasn't
confirmed yet; the ATR stop-loss (see risk.py) is what protects you
during that gap, not this signal
Why state-based: an earlier version only fired BUY/SELL on the exact bar
a crossover occurred, gated by ADX on that same bar. If ADX was still
climbing toward the threshold when the crossover happened (a common
pattern — price often moves before ADX confirms it's a real trend), the
strategy would correctly hold off, but then never re-evaluate once ADX
caught up days later, because the "crossing" moment had already passed.
Checking state every bar instead of only the crossing bar fixes that on
both entry and exit.
Why ADX gates the exit too: an intermediate version of this fix removed
the ADX gate from the exit only, reasoning that the ATR stop-loss already
covers downside risk. Backtesting showed this let the strategy bail on
minor pullbacks within an otherwise-healthy uptrend and re-buy shortly
after, generating more whipsaw losses than it prevented (confirmed on
both ACN and AAPL — trade count roughly doubled and performance vs.
buy-and-hold got worse on both). Gating the exit on ADX again, same as
the original design, fixed that while keeping the entry-side bug fix.
This is a scaffold to prove the pipeline works and is a step up from a
bare crossover — it is not a strategy proven to be profitable. Keep
iterating and validating on out-of-sample data before trusting it with
real money.
"""
import pandas as pd
import config
def add_moving_averages(df: pd.DataFrame,
short_window: int = config.SHORT_WINDOW,
long_window: int = config.LONG_WINDOW) -> pd.DataFrame:
df = df.copy()
df["sma_short"] = df["close"].rolling(window=short_window).mean()
df["sma_long"] = df["close"].rolling(window=long_window).mean()
return df
def add_adx(df: pd.DataFrame, period: int = config.ADX_PERIOD) -> pd.DataFrame:
"""
Adds an 'adx' column measuring trend strength (0-100, Wilder's method).
Above ~25 is generally considered "trending"; below ~20 is choppy/flat.
Requires 'high', 'low', 'close' columns.
"""
df = df.copy()
high, low, close = df["high"], df["low"], df["close"]
prev_close = close.shift(1)
tr = pd.concat([
high - low,
(high - prev_close).abs(),
(low - prev_close).abs(),
], axis=1).max(axis=1)
up_move = high.diff()
down_move = -low.diff()
plus_dm = pd.Series(0.0, index=df.index)
minus_dm = pd.Series(0.0, index=df.index)
plus_dm[(up_move > down_move) & (up_move > 0)] = up_move[(up_move > down_move) & (up_move > 0)]
minus_dm[(down_move > up_move) & (down_move > 0)] = down_move[(down_move > up_move) & (down_move > 0)]
# Wilder's smoothing (equivalent to an EMA with alpha = 1/period)
atr = tr.ewm(alpha=1 / period, adjust=False).mean()
df["atr"] = atr # exposed for volatility-based stop-loss sizing (see risk.py)
plus_di = 100 * plus_dm.ewm(alpha=1 / period, adjust=False).mean() / atr
minus_di = 100 * minus_dm.ewm(alpha=1 / period, adjust=False).mean() / atr
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di)
df["adx"] = dx.ewm(alpha=1 / period, adjust=False).mean()
return df
def generate_signal(df: pd.DataFrame, adx_threshold: float = config.ADX_THRESHOLD) -> str:
"""
Given a DataFrame with sma_short/sma_long/adx columns (most recent row
last), return 'BUY', 'SELL', or 'HOLD' based on the CURRENT state —
not just whether a crossover happened on this exact bar. See module
docstring for why this matters.
"""
if len(df) < 1:
return "HOLD"
curr = df.iloc[-1]
if pd.isna(curr["sma_short"]) or pd.isna(curr["sma_long"]) or pd.isna(curr["adx"]):
return "HOLD"
above = curr["sma_short"] > curr["sma_long"]
trending = curr["adx"] >= adx_threshold
# ADX gates BOTH entry and exit — not just entry. An early version of
# this fix removed the gate on exit (reasoning that the ATR stop-loss
# covers downside risk anyway), but backtesting showed that let the
# strategy exit on minor pullbacks within a still-healthy uptrend and
# re-buy shortly after, racking up whipsaw losses on both ACN and
# AAPL. Requiring ADX confirmation on the way out too — same as the
# original design — cut that down significantly.
if above and trending:
return "BUY"
if not above and trending:
return "SELL"
return "HOLD"
def generate_signals_series(df: pd.DataFrame,
adx_threshold: float = config.ADX_THRESHOLD) -> pd.Series:
"""Vectorized version used by the backtester — returns a signal per row."""
df = df.copy()
above = df["sma_short"] > df["sma_long"]
trending = df["adx"] >= adx_threshold
signals = pd.Series("HOLD", index=df.index)
signals[above & trending] = "BUY"
signals[(~above) & trending] = "SELL"
return signals
A quick note on the design here, since it’s easy to get subtly wrong: signals are evaluated as a current state (is the short average above the long average right now, and does ADX confirm a trend right now) rather than only firing on the exact bar a crossover happens. An earlier version of this logic only fired on the crossing bar itself — which sounds equivalent but isn’t: if ADX takes a few days to confirm a trend after the crossover already happened, an event-based check misses the entry entirely, since the “crossing” moment has already passed by the time confirmation arrives. Checking the state fresh on every bar avoids that.
risk.py — position sizing and stop-loss
"""
Basic risk management: position sizing and stop-loss checks.
Deliberately simple — extend with correlation checks across symbols, max
daily loss limits, etc. as you go.
"""
import config
def position_size(equity: float, price: float,
max_position_pct: float = config.MAX_POSITION_PCT) -> int:
"""
Return the number of whole shares to buy given account equity, current
price, and the max fraction of equity to risk on this position.
"""
if price <= 0:
return 0
dollar_amount = equity * max_position_pct
shares = int(dollar_amount // price)
return max(shares, 0)
def stop_loss_price(entry_price: float, entry_atr: float,
multiplier: float = config.ATR_STOP_MULTIPLIER) -> float:
"""
Return the price at which a long position should be stopped out, based
on the asset's own volatility (ATR) at the time of entry rather than a
flat percentage. A volatile symbol gets a wider stop; a calm one gets
a tighter stop — both scaled the same way.
"""
return entry_price - (multiplier * entry_atr)
def stop_loss_triggered(entry_price: float, current_price: float,
entry_atr: float,
multiplier: float = config.ATR_STOP_MULTIPLIER) -> bool:
"""True if current_price has fallen through the ATR-based stop level."""
if entry_price <= 0 or entry_atr <= 0:
return False
return current_price <= stop_loss_price(entry_price, entry_atr, multiplier)
The stop-loss uses ATR (Average True Range) instead of a flat percentage. Here’s why that matters: a flat 2% stop sounds reasonable until you apply it to a volatile stock — a 2% intraday swing can be completely normal noise for a volatile name, so a flat stop just gets you shaken out repeatedly by nothing. Scaling the stop to each symbol’s own recent volatility means a calm stock gets a tight stop and a volatile one gets appropriately more room.
executor.py — talking to Alpaca
"""
Handles submitting orders to Alpaca's paper trading account and querying
account/position state.
"""
from alpaca.trading.client import TradingClient
from alpaca.trading.requests import MarketOrderRequest, GetOrdersRequest
from alpaca.trading.enums import OrderSide, TimeInForce, QueryOrderStatus
import config
class Executor:
def __init__(self):
self.client = TradingClient(
config.ALPACA_API_KEY,
config.ALPACA_SECRET_KEY,
paper=config.PAPER_TRADING,
)
def get_equity(self) -> float:
account = self.client.get_account()
return float(account.equity)
def get_position_qty(self, symbol: str) -> float:
try:
position = self.client.get_open_position(symbol)
return float(position.qty)
except Exception as e:
# Alpaca raises an exception when there's genuinely no open
# position for this symbol, which is the expected/common case
# — but the same broad except also swallows real errors (auth
# failures, rate limits, network issues), silently treating
# them as "no position" too. Print so a persistent real error
# is visible in the console instead of masquerading as "flat".
msg = str(e).lower()
if "position does not exist" not in msg and "404" not in msg:
print(f"[executor] WARNING: unexpected error checking position "
f"for {symbol}: {e}")
return 0.0
def get_position_entry_price(self, symbol: str) -> float:
"""Returns the average entry price of the open position, or 0.0 if none."""
try:
position = self.client.get_open_position(symbol)
return float(position.avg_entry_price)
except Exception:
return 0.0
def has_open_order(self, symbol: str) -> bool:
"""
True if there's already a pending (unfilled) order for this symbol
— checked BEFORE submitting a new one, so the bot doesn't stack up
duplicate orders when a previous order hasn't filled yet (e.g. it
was submitted outside market hours and is still queued).
"""
try:
request = GetOrdersRequest(status=QueryOrderStatus.OPEN, symbols=[symbol])
open_orders = self.client.get_orders(filter=request)
return len(open_orders) > 0
except Exception as e:
print(f"[executor] WARNING: could not check open orders for "
f"{symbol}: {e}. Assuming none, to avoid blocking trading "
f"entirely on a transient API error.")
return False
def submit_market_order(self, symbol: str, qty: float, side: str):
"""side: 'BUY' or 'SELL'"""
order_side = OrderSide.BUY if side == "BUY" else OrderSide.SELL
order_request = MarketOrderRequest(
symbol=symbol,
qty=qty,
side=order_side,
time_in_force=TimeInForce.DAY,
)
order = self.client.submit_order(order_request)
return order
Two things worth calling out:
has_open_order()exists specifically to prevent duplicate orders. If you submit an order outside market hours, it queues rather than filling instantly — without checking for pending orders first, a polling loop can submit several duplicate orders before the first one ever clears, and they can all fill at once when the market opens. Ask me how I know.get_position_qty()‘s exception handling is intentionally narrow — it only treats “no position exists” as the expected case, and prints a warning for anything else, so a real API error doesn’t get silently mistaken for “you have no position.”
trade_logger.py — recording everything
"""
Logs every trade decision (and skipped/held bars, optionally) to a CSV so
you can review performance later without re-querying the broker.
Log filenames are per-symbol (e.g. trade_log_ACN.csv, equity_log_SPY.csv)
so multiple bot instances running from the same project directory don't
overwrite each other's data.
"""
import csv
import os
from datetime import datetime
_FIELDS = ["timestamp", "symbol", "action", "qty", "price", "note"]
_EQUITY_FIELDS = ["timestamp", "symbol", "equity", "price", "position_qty"]
_INDICATOR_FIELDS = ["timestamp", "symbol", "sma_short", "sma_long", "adx", "gap_pct"]
def trade_log_path(symbol: str) -> str:
return f"trade_log_{symbol}.csv"
def equity_log_path(symbol: str) -> str:
return f"equity_log_{symbol}.csv"
def indicator_log_path(symbol: str) -> str:
return f"indicator_log_{symbol}.csv"
def log_trade(symbol: str, action: str, qty: float, price: float, note: str = ""):
path = trade_log_path(symbol)
file_exists = os.path.isfile(path)
with open(path, "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=_FIELDS)
if not file_exists:
writer.writeheader()
writer.writerow({
"timestamp": datetime.now().isoformat(timespec="seconds"),
"symbol": symbol,
"action": action,
"qty": qty,
"price": price,
"note": note,
})
def log_equity_snapshot(symbol: str, equity: float, price: float, position_qty: float):
"""
Records account equity + the symbol's price at a point in time, even on
days with no trade. This is what lets you plot an equity curve and
compare it against buy-and-hold later — trade_log.csv alone can't do
that since it only has rows on days something was bought/sold.
"""
path = equity_log_path(symbol)
file_exists = os.path.isfile(path)
with open(path, "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=_EQUITY_FIELDS)
if not file_exists:
writer.writeheader()
writer.writerow({
"timestamp": datetime.now().isoformat(timespec="seconds"),
"symbol": symbol,
"equity": equity,
"price": price,
"position_qty": position_qty,
})
def log_indicator_snapshot(symbol: str, sma_short: float, sma_long: float, adx: float):
"""
Records the strategy's current SMA20/SMA50/ADX state, once per day, so
the HTML dashboard can show how close the strategy is to a signal
without needing to watch the console output directly.
"""
path = indicator_log_path(symbol)
file_exists = os.path.isfile(path)
gap_pct = (sma_short - sma_long) / sma_long * 100 if sma_long else 0.0
with open(path, "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=_INDICATOR_FIELDS)
if not file_exists:
writer.writeheader()
writer.writerow({
"timestamp": datetime.now().isoformat(timespec="seconds"),
"symbol": symbol,
"sma_short": sma_short,
"sma_long": sma_long,
"adx": adx,
"gap_pct": gap_pct,
})
This logs to per-symbol CSV files (trade_log_ACN.csv, equity_log_SPY.csv, etc.) so you can run multiple symbols from the same project directory without their logs overwriting each other.
Step 6: Backtest before running anything live
This is the step I’d encourage you not to skip, whatever strategy you end up using. Here’s the backtesting script:
backtest.py
"""
Backtests the SMA crossover strategy against historical daily bars.
No orders are submitted — this only uses Alpaca for historical DATA.
Usage:
python backtest.py --symbol AAPL --days 180 --capital 10000
python backtest.py --symbol AAPL --days 730 --short 10 --long 30
python backtest.py --symbol AAPL --days 730 --adx-threshold 20
"""
import argparse
import pandas as pd
import config
import risk
from data_feed import DataFeed
from strategy import add_moving_averages, add_adx, generate_signals_series
def run_backtest(symbol: str, days: int, starting_capital: float,
short_window: int, long_window: int, adx_threshold: float):
feed = DataFeed()
df = feed.get_historical_bars(symbol, days=days)
df = add_moving_averages(df, short_window=short_window, long_window=long_window)
df = add_adx(df)
df["signal"] = generate_signals_series(df, adx_threshold=adx_threshold)
cash = starting_capital
shares = 0
entry_price = 0.0
entry_atr = 0.0
trades = []
stop_loss_exits = 0
for ts, row in df.iterrows():
price = row["close"]
signal = row["signal"]
# Stop-loss check runs FIRST and independent of the strategy signal —
# protects against a position drifting down without ever producing
# a clean bearish crossover. Uses ATR-at-entry so the stop distance
# scales with this symbol's own volatility instead of a flat %.
if shares > 0 and risk.stop_loss_triggered(entry_price, price, entry_atr):
proceeds = shares * price
cash += proceeds
pnl = (price - entry_price) * shares
trades.append((ts, "SELL (stop-loss)", shares, price, pnl))
shares = 0
entry_price = 0.0
entry_atr = 0.0
stop_loss_exits += 1
continue
if signal == "BUY" and shares == 0:
shares = int(cash // price)
if shares > 0:
cost = shares * price
cash -= cost
entry_price = price
entry_atr = row["atr"]
trades.append((ts, "BUY", shares, price))
elif signal == "SELL" and shares > 0:
proceeds = shares * price
cash += proceeds
pnl = (price - entry_price) * shares
trades.append((ts, "SELL", shares, price, pnl))
shares = 0
entry_price = 0.0
entry_atr = 0.0
# Close any open position at the last price for a fair final comparison
final_price = df["close"].iloc[-1]
if shares > 0:
cash += shares * final_price
pnl = (final_price - entry_price) * shares
trades.append((df.index[-1], "SELL (close)", shares, final_price, pnl))
shares = 0
total_return_pct = (cash - starting_capital) / starting_capital * 100
sell_trades = [t for t in trades if t[1].startswith("SELL")]
wins = [t for t in sell_trades if t[4] > 0]
print(f"\n--- Backtest: {symbol} over last {days} days "
f"(SMA {short_window}/{long_window}, ADX>={adx_threshold}) ---")
print(f"Starting capital: ${starting_capital:,.2f}")
print(f"Ending capital: ${cash:,.2f}")
print(f"Total return: {total_return_pct:.2f}%")
print(f"Total trades: {len(trades)}")
print(f"Stop-loss exits: {stop_loss_exits}")
if sell_trades:
win_rate = len(wins) / len(sell_trades) * 100
print(f"Win rate: {win_rate:.1f}% ({len(wins)}/{len(sell_trades)})")
trending_pct = (df["adx"] >= adx_threshold).mean() * 100
print(f"Time trending: {trending_pct:.1f}% of bars had ADX >= {adx_threshold}")
# Buy-and-hold benchmark: what if you'd just bought on day 1 and held?
first_valid_price = df["close"].iloc[0]
last_price = df["close"].iloc[-1]
bh_shares = int(starting_capital // first_valid_price)
bh_final_value = bh_shares * last_price + (starting_capital - bh_shares * first_valid_price)
bh_return_pct = (bh_final_value - starting_capital) / starting_capital * 100
print(f"\nBuy & hold benchmark ({symbol}, same period): {bh_return_pct:.2f}%")
print(f"Strategy vs. buy & hold: {total_return_pct - bh_return_pct:+.2f} percentage points")
print("\nTrade log:")
for t in trades:
print(t)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--symbol", default="AAPL")
parser.add_argument("--days", type=int, default=180)
parser.add_argument("--capital", type=float, default=10000)
parser.add_argument("--short", type=int, default=config.SHORT_WINDOW,
help="Short SMA window (default from config.py)")
parser.add_argument("--long", type=int, default=config.LONG_WINDOW,
help="Long SMA window (default from config.py)")
parser.add_argument("--adx-threshold", type=float, default=config.ADX_THRESHOLD,
help="Minimum ADX to allow a trade (default from config.py)")
args = parser.parse_args()
run_backtest(args.symbol, args.days, args.capital,
args.short, args.long, args.adx_threshold)
Run it like this:
python3 backtest.py --symbol AAPL --days 730
The most important line in that output is the buy-and-hold comparison. A strategy can show a positive return and still be a bad idea — the only way to know is to compare it against what you’d have gotten just buying and holding the same stock over the same period. Don’t skip this comparison; it’s the single most useful sanity check in the whole project.
You can also tune the strategy’s parameters directly from the command line without touching the code:
python3 backtest.py --symbol AAPL --days 730 --short 10 --long 30
python3 backtest.py --symbol AAPL --days 730 --adx-threshold 20
Step 7: Run the live paper-trading loop
Once you’re comfortable with the backtest results, here’s the live trading script:
main.py
"""
Live (paper) trading loop.
Polls the latest bars on an interval, feeds them to the strategy, and
submits paper orders through Alpaca when a signal fires.
Usage:
python main.py --symbol AAPL --qty 1
python main.py --symbol AAPL --qty 1 --interval 300
"""
import argparse
import time
from datetime import datetime, timezone
from data_feed import DataFeed
from strategy import add_moving_averages, add_adx, generate_signal
from executor import Executor
from trade_logger import log_trade, log_equity_snapshot, log_indicator_snapshot
import risk
def _drop_incomplete_today_bar(df):
"""
The most recent daily bar keeps changing while the market is open — it's
not a finalized close yet. Using it directly can cause signals to
flicker mid-day based on data that hasn't settled. Drop it so every
decision is based only on confirmed, closed daily bars.
"""
today = datetime.now(timezone.utc).date()
if df.index[-1].date() == today:
return df.iloc[:-1]
return df
def run(symbol: str, interval: int, use_risk_sizing: bool, fixed_qty: int):
feed = DataFeed()
executor = Executor()
# Tracks the ATR at the moment we entered the current position, so the
# stop-loss distance reflects the symbol's volatility at entry time
# rather than a flat percentage. NOTE: this only persists while this
# process keeps running — if you restart main.py while holding an open
# position, entry_atr resets to None and the stop-loss falls back to
# using the *current* ATR as a reasonable approximation until the
# position is closed and reopened.
entry_atr = None
last_snapshot_date = None # tracks whether we've logged equity today yet
print(f"[{datetime.now()}] Starting paper trading loop for {symbol} "
f"(interval={interval}s). Ctrl+C to stop.")
while True:
try:
# Pull enough history to compute both moving averages
df = feed.get_historical_bars(symbol, days=90)
df = _drop_incomplete_today_bar(df)
df = add_moving_averages(df)
df = add_adx(df)
signal = generate_signal(df)
last_close = df["close"].iloc[-1]
current_atr = df["atr"].iloc[-1]
sma_short = df["sma_short"].iloc[-1]
sma_long = df["sma_long"].iloc[-1]
current_adx = df["adx"].iloc[-1]
# Signals above are based on confirmed closes only (avoids
# flickering on an incomplete bar). But the stop-loss should
# react to the live price during the day, not wait for the
# close — otherwise a fast intraday drop wouldn't trigger it
# until tomorrow, defeating the point of a stop-loss.
latest_bar = feed.get_latest_bar(symbol)
live_price = latest_bar.close
current_qty = executor.get_position_qty(symbol)
# Log an equity snapshot once per day, regardless of whether a
# trade happens — this is what lets you plot an equity curve
# later (trade_log.csv alone only has rows on trade days).
today = datetime.now(timezone.utc).date()
if last_snapshot_date != today:
equity = executor.get_equity()
log_equity_snapshot(symbol, equity, live_price, current_qty)
log_indicator_snapshot(symbol, sma_short, sma_long, current_adx)
last_snapshot_date = today
# --- Indicator visibility: shows exactly how close we are to a
# signal firing, so you don't have to guess from outside data.
gap_pct = (sma_short - sma_long) / sma_long * 100 if sma_long else 0
cross_state = "ABOVE" if sma_short > sma_long else "below"
adx_state = "TRENDING" if current_adx >= 25 else "not trending"
print(f"[{datetime.now()}] {symbol} last_close={last_close:.2f} "
f"live={live_price:.2f} | SMA20={sma_short:.2f} "
f"SMA50={sma_long:.2f} ({cross_state}, gap={gap_pct:+.2f}%) "
f"| ADX={current_adx:.1f} ({adx_state}) | position={current_qty}")
# Stop-loss check runs FIRST and independent of the strategy
# signal — protects against a position drifting down without
# ever producing a clean bearish crossover.
if current_qty > 0:
entry_price = executor.get_position_entry_price(symbol)
atr_for_stop = entry_atr if entry_atr is not None else current_atr
if risk.stop_loss_triggered(entry_price, live_price, atr_for_stop):
if executor.has_open_order(symbol):
print(f"[{datetime.now()}] STOP-LOSS triggered, but "
f"an order for {symbol} is already pending — "
f"skipping to avoid a duplicate.")
else:
executor.submit_market_order(symbol, current_qty, "SELL")
log_trade(symbol, "SELL", current_qty, live_price, note="stop-loss")
print(f"[{datetime.now()}] STOP-LOSS: SELL {current_qty} {symbol} "
f"@ ~{live_price:.2f} (entry was {entry_price:.2f})")
entry_atr = None
time.sleep(interval)
continue
if signal == "BUY" and current_qty == 0:
if executor.has_open_order(symbol):
print(f"[{datetime.now()}] BUY signal, but an order for "
f"{symbol} is already pending (likely queued "
f"outside market hours) — skipping to avoid a "
f"duplicate.")
elif use_risk_sizing:
equity = executor.get_equity()
qty = risk.position_size(equity, live_price)
if qty > 0:
executor.submit_market_order(symbol, qty, "BUY")
log_trade(symbol, "BUY", qty, live_price)
entry_atr = current_atr
print(f"[{datetime.now()}] BUY {qty} {symbol} @ ~{live_price:.2f}")
else:
print(f"[{datetime.now()}] BUY signal but sized qty is 0 — skipping.")
else:
qty = fixed_qty
executor.submit_market_order(symbol, qty, "BUY")
log_trade(symbol, "BUY", qty, live_price)
entry_atr = current_atr
print(f"[{datetime.now()}] BUY {qty} {symbol} @ ~{live_price:.2f}")
elif signal == "SELL" and current_qty > 0:
if executor.has_open_order(symbol):
print(f"[{datetime.now()}] SELL signal, but an order for "
f"{symbol} is already pending — skipping to avoid "
f"a duplicate.")
else:
executor.submit_market_order(symbol, current_qty, "SELL")
log_trade(symbol, "SELL", current_qty, live_price)
entry_atr = None
print(f"[{datetime.now()}] SELL {current_qty} {symbol} @ ~{live_price:.2f}")
except Exception as e:
print(f"[{datetime.now()}] ERROR: {e}")
time.sleep(interval)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--symbol", default="AAPL")
parser.add_argument("--qty", type=int, default=1,
help="Fixed share quantity (ignored if --risk-sizing is set)")
parser.add_argument("--interval", type=int, default=900,
help="Seconds between checks (default 900 = 15 min; "
"signals only change once/day, but a shorter "
"interval lets the stop-loss react intraday)")
parser.add_argument("--risk-sizing", action="store_true",
help="Size positions using risk.position_size() instead of --qty")
args = parser.parse_args()
run(args.symbol, args.interval, args.risk_sizing, args.qty)
Run it with:
python3 main.py --symbol AAPL --qty 1
This polls every 15 minutes by default, checks the strategy’s signal, and submits paper orders through Alpaca when conditions are met. Each cycle prints the current indicator values (moving averages, ADX, current position) so you can watch it work in real time instead of guessing what it’s doing.
Step 8: Keep it running with tmux
SSH sessions die, laptops sleep — you don’t want the bot to stop just because your terminal disconnected. tmux solves this:
tmux new -s tradingbot
python3 main.py --symbol AAPL --qty 1
Detach with Ctrl+B then D (press and release Ctrl+B, then press D separately — it’s a two-step sequence, not a single chord). The bot keeps running in the background. Reattach anytime with:
tmux attach -t tradingbot
Step 9: Check in on it periodically
A few habits worth building once this is running unattended:
- Check the Alpaca dashboard directly, not just the bot’s own logs — if there’s ever a bug in the logging, the dashboard is the source of truth.
- Redirect output to a file if running detached, so you have something to review later:
python3 main.py --symbol AAPL --qty 1 > bot.log 2>&1 & - Watch for repeated
ERROR:lines in the console — the loop catches exceptions and keeps running (good for uptime), but that also means a persistent problem like a bad API key or rate limiting could fail silently every cycle unless you’re watching for it.
A couple of hard-earned lessons
Don’t gate your stop-loss on the same trend filter as your entry. I initially removed the ADX confirmation from the exit condition, reasoning the ATR stop already handled downside risk. Backtesting showed this let the strategy bail on ordinary pullbacks within an otherwise healthy trend, then re-buy shortly after — generating more whipsaw losses than it prevented. Requiring trend confirmation on both entry and exit, symmetrically, fixed it.
Always check for pending orders, not just filled positions, before submitting a new one. This is the bug that cost me fourteen unintended paper trades in one morning. If an order is submitted while the market is closed, it sits queued rather than executing — and a polling loop that only checks “do I already hold a position” has no way of knowing a previous order is still in flight. has_open_order() in executor.py above is the fix.
Where to go from here
This strategy is intentionally simple — a moving-average crossover isn’t going to be anyone’s path to reliably beating the market. The value of this project isn’t the strategy itself; it’s the scaffolding around it: a clean separation between data, signal generation, risk, and execution, a backtester you actually trust, and a live loop that fails safely. Swap in your own signal logic once you’re comfortable with the pipeline — generate_signal() in strategy.py just needs to keep returning BUY, SELL, or HOLD, and everything downstream keeps working.
Up next: Part 3 adds a live, self-updating performance dashboard — a dark, terminal-style web page comparing the strategy against buy-and-hold in real time, served over Apache on the same server.