Skip to content

Add adaptive day trading bot with risk controls #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bofhus wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add adaptive day trading bot with risk controls #1

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
303 changes: 219 additions & 84 deletions bot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import json
import os
import sys
import time
import sqlite3
import sys
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Dict, Iterable, List

import numpy as np
import pandas as pd
Expand All @@ -23,15 +24,25 @@
# -----------------------------
@dataclass
class Config:
symbol: str = "SPY" # Byt till valfri US-aktie/ETF
symbols: List[str] = None # Lista med tickers att scanna
timeframe: TimeFrame = TimeFrame.Minute
lookback_minutes: int = 3000 # ~2 handelsdagar av 1-min bars (räcker för SMA50)
sma_fast: int = 20
sma_slow: int = 50
max_position_pct: float = 0.25 # max 25% av equity i denna symbol
min_equity_usd: float = 100.0 # kill switch om equity “försvinner”
lookback_minutes: int = 800 # ~en handelsdag av 1-min bars
momentum_fast: int = 15
momentum_slow: int = 60
evaluation_horizon_minutes: int = 60 # Efter hur lång tid vi utvärderar signalerna
max_position_pct: float = 0.15 # max 15% av equity i en symbol
max_daily_budget_usd: float = 500.0 # max USD att köpa för per dag
fee_pct: float = 0.0005 # antagen courtage/avgift
stop_loss_pct: float = 0.01 # säljer om priset faller X% från entry
take_profit_pct: float = 0.02 # kan sälja om vinsten passerar denna nivå
min_equity_usd: float = 100.0 # kill switch om equity “försvinner”
db_path: str = "bot_state.sqlite3"

def __post_init__(self):
if self.symbols is None:
# Default-universum av stora, likvida US-aktier/ETF:er
self.symbols = ["SPY", "QQQ", "AAPL", "MSFT", "AMZN", "TSLA"]


def setup_logging() -> None:
logging.basicConfig(
Expand All @@ -53,12 +64,14 @@ def get_env_bool(name: str, default: bool = False) -> bool:
# -----------------------------
def init_db(db_path: str) -> None:
with sqlite3.connect(db_path) as conn:
conn.execute("""
conn.execute(
"""
CREATE TABLE IF NOT EXISTS state (
k TEXT PRIMARY KEY,
v TEXT NOT NULL
)
""")
"""
)
conn.commit()


Expand All @@ -75,37 +88,65 @@ def state_set(db_path: str, key: str, value: str) -> None:
conn.commit()


def state_get_json(db_path: str, key: str, default):
raw = state_get(db_path, key, "")
if not raw:
return default
try:
return json.loads(raw)
except json.JSONDecodeError:
logging.warning("Kunde inte tolka JSON för %s, reset.", key)
return default


def state_set_json(db_path: str, key: str, value) -> None:
state_set(db_path, key, json.dumps(value))


# -----------------------------
# Strategy
# -----------------------------
def compute_sma(series: pd.Series, window: int) -> pd.Series:
return series.rolling(window=window).mean()


def generate_signal(df: pd.DataFrame, fast: int, slow: int) -> str | None:
def momentum_score(df: pd.DataFrame, fast: int, slow: int) -> float:
"""
Returnerar "buy", "sell" eller None.
SMA-cross: fast korsar över slow => buy, under => sell.
Enkel momentum-score baserat på kort vs lång avkastning och trend.
Returnerar högre värden för positivt momentum.
"""
close = df["close"]
sma_f = compute_sma(close, fast)
sma_s = compute_sma(close, slow)

if len(df) < slow + 2:
return None

# senaste två punkter för kors
f_prev, f_now = sma_f.iloc[-2], sma_f.iloc[-1]
s_prev, s_now = sma_s.iloc[-2], sma_s.iloc[-1]
if len(df) < slow + 5:
return float("nan")

if np.isnan([f_prev, f_now, s_prev, s_now]).any():
return None

if f_prev <= s_prev and f_now > s_now:
return "buy"
if f_prev >= s_prev and f_now < s_now:
return "sell"
return None
close = df["close"]
ret = close.pct_change()
fast_ret = ret.tail(fast).mean()
slow_ret = ret.tail(slow).mean()
trend = close.iloc[-1] / close.rolling(window=slow).mean().iloc[-1] - 1
vol_penalty = ret.tail(slow).std()

score = 0.5 * fast_ret + 0.3 * (fast_ret - slow_ret) + 0.2 * trend
if vol_penalty > 0:
score -= vol_penalty * 0.1
return float(score)


def ranked_candidates(
data: Dict[str, pd.DataFrame],
cfg: Config,
skills: Dict[str, float],
max_candidates: int = 3,
) -> List[str]:
ranked: List[tuple[str, float]] = []
for symbol, df in data.items():
sc = momentum_score(df, cfg.momentum_fast, cfg.momentum_slow)
if np.isnan(sc):
continue
skill_bonus = skills.get(symbol, 0.0)
ranked.append((symbol, sc + 0.1 * skill_bonus))

ranked.sort(key=lambda x: x[1], reverse=True)
return [s for s, _ in ranked[:max_candidates] if _ > 0]


# -----------------------------
Expand All @@ -117,42 +158,50 @@ def get_clients(api_key: str, secret_key: str, paper: bool):
return trading, data


def fetch_bars(data_client: StockHistoricalDataClient, cfg: Config) -> pd.DataFrame:
def fetch_bars(
data_client: StockHistoricalDataClient, cfg: Config, symbols: Iterable[str]
) -> Dict[str, pd.DataFrame]:
end = datetime.now(timezone.utc)
start = end - timedelta(minutes=cfg.lookback_minutes)

req = StockBarsRequest(
symbol_or_symbols=cfg.symbol,
symbol_or_symbols=list(symbols),
timeframe=cfg.timeframe,
start=start,
end=end
end=end,
)

bars = data_client.get_stock_bars(req)
df = bars.df
result: Dict[str, pd.DataFrame] = {}

# alpaca-py returnerar MultiIndex (symbol, timestamp) om flera symboler
if isinstance(df.index, pd.MultiIndex):
df = df.xs(cfg.symbol)
if not isinstance(df.index, pd.MultiIndex):
df = df.sort_index().rename(columns=str.lower)
result[req.symbol_or_symbols] = df
return result

df = df.sort_index()
# standardisera kolumnnamn
df = df.rename(columns=str.lower)
return df
for symbol in req.symbol_or_symbols:
sub = df.xs(symbol)
sub = sub.sort_index().rename(columns=str.lower)
result[symbol] = sub

return result


def get_equity(trading: TradingClient) -> float:
acct = trading.get_account()
return float(acct.equity)


def get_position_qty(trading: TradingClient, symbol: str) -> int:
positions = trading.get_all_positions()
for p in positions:
if p.symbol == symbol:
# qty kan vara str
return int(float(p.qty))
return 0
def get_positions(trading: TradingClient) -> Dict[str, dict]:
positions = {}
for p in trading.get_all_positions():
positions[p.symbol] = {
"qty": int(float(p.qty)),
"avg_entry": float(p.avg_entry_price),
"market_value": float(p.market_value),
}
return positions


def target_qty_for_buy(equity: float, last_price: float, max_pct: float) -> int:
Expand All @@ -176,6 +225,115 @@ def submit_market_order(trading: TradingClient, symbol: str, qty: int, side: str
logging.info("Skickade %s market order: %s x %d", side.upper(), symbol, qty)


def reset_daily_budget(db_path: str) -> float:
today = datetime.now(timezone.utc).date().isoformat()
saved_date = state_get(db_path, "daily_date", "")
if saved_date != today:
state_set(db_path, "daily_date", today)
state_set(db_path, "daily_spent", "0.0")
spent = float(state_get(db_path, "daily_spent", "0.0"))
return spent


def add_daily_spent(db_path: str, amount: float) -> None:
spent = float(state_get(db_path, "daily_spent", "0.0"))
state_set(db_path, "daily_spent", str(spent + amount))


def update_skills_from_pending(db_path: str, prices: Dict[str, float], cfg: Config) -> Dict[str, float]:
skills: Dict[str, float] = state_get_json(db_path, "skills", {})
pending = state_get_json(db_path, "pending_eval", {})
now = datetime.now(timezone.utc)
horizon = timedelta(minutes=cfg.evaluation_horizon_minutes)
updated_pending = {}

for symbol, payload in pending.items():
ts = datetime.fromisoformat(payload["ts"])
entry_price = payload["entry"]
if now - ts < horizon:
updated_pending[symbol] = payload
continue
if symbol not in prices:
updated_pending[symbol] = payload
continue

ret = (prices[symbol] - entry_price) / entry_price
skills[symbol] = skills.get(symbol, 0.0) + float(ret)
logging.info("Utvärderade %s: avkastning %.2f%% -> ny skill %.4f", symbol, ret * 100, skills[symbol])

state_set_json(db_path, "skills", skills)
state_set_json(db_path, "pending_eval", updated_pending)
return skills


def record_pending_eval(db_path: str, symbol: str, entry_price: float) -> None:
pending = state_get_json(db_path, "pending_eval", {})
pending[symbol] = {
"entry": entry_price,
"ts": datetime.now(timezone.utc).isoformat(),
}
state_set_json(db_path, "pending_eval", pending)


def enforce_stop_rules(trading: TradingClient, prices: Dict[str, float], positions: Dict[str, dict], cfg: Config) -> None:
for symbol, pos in positions.items():
if symbol not in prices:
continue
qty = pos["qty"]
if qty <= 0:
continue
entry = pos["avg_entry"]
last = prices[symbol]
change = (last - entry) / entry
if change <= -cfg.stop_loss_pct:
logging.info("Stop loss triggat för %s: %.2f%%", symbol, change * 100)
submit_market_order(trading, symbol, qty, "sell")
elif change >= cfg.take_profit_pct:
logging.info("Tar vinst för %s: %.2f%%", symbol, change * 100)
submit_market_order(trading, symbol, qty, "sell")


def buy_candidates(
trading: TradingClient,
candidates: Iterable[str],
prices: Dict[str, float],
positions: Dict[str, dict],
equity: float,
cfg: Config,
db_path: str,
) -> None:
spent_today = reset_daily_budget(db_path)
remaining_budget = max(cfg.max_daily_budget_usd - spent_today, 0)

for symbol in candidates:
if remaining_budget <= 0:
logging.info("Dagsbudget förbrukad. Bryter köp.")
break
if positions.get(symbol, {}).get("qty", 0) > 0:
logging.info("Redan long %s, hoppar över köp.", symbol)
continue

price = prices.get(symbol)
if price is None or price <= 0:
continue

max_for_symbol = min(equity * cfg.max_position_pct, remaining_budget)
if max_for_symbol <= 0:
continue

qty = int((max_for_symbol / (1 + cfg.fee_pct)) // price)
if qty <= 0:
logging.info("Budget otillräcklig för %s.", symbol)
continue

submit_market_order(trading, symbol, qty, "buy")
cost_estimate = price * qty * (1 + cfg.fee_pct)
add_daily_spent(db_path, cost_estimate)
remaining_budget -= cost_estimate
record_pending_eval(db_path, symbol, price)
logging.info("Köpte %s x %d för ca %.2f USD (avgift %.4f%%).", symbol, qty, cost_estimate, cfg.fee_pct * 100)


# -----------------------------
# Main
# -----------------------------
Expand All @@ -201,48 +359,25 @@ def main():
logging.error("Equity (%.2f) under min (%.2f). Avbryter.", equity, cfg.min_equity_usd)
return

# Hämta data och signal
df = fetch_bars(data, cfg)
if df.empty:
# Hämta data för alla symboler
data_by_symbol = fetch_bars(data, cfg, cfg.symbols)
if not data_by_symbol:
logging.warning("Ingen data returnerad. Avbryter.")
return

last_price = float(df["close"].iloc[-1])
signal = generate_signal(df, cfg.sma_fast, cfg.sma_slow)
logging.info("Symbol=%s last=%.2f signal=%s", cfg.symbol, last_price, signal)
prices = {s: float(df["close"].iloc[-1]) for s, df in data_by_symbol.items() if not df.empty}

if signal is None:
return
# Självförbättring: uppdatera skills baserat på tidigare prediktioner
skills = update_skills_from_pending(cfg.db_path, prices, cfg)

# Anti-spam: agera bara om signalen ändrats sen sist
last_signal = state_get(cfg.db_path, "last_signal", "")
if signal == last_signal:
logging.info("Signal oförändrad (%s). Ingen ny order.", signal)
return
# Riskhantering för öppna positioner
positions = get_positions(trading)
enforce_stop_rules(trading, prices, positions, cfg)

# Nuvarande position
pos_qty = get_position_qty(trading, cfg.symbol)
logging.info("Nuvarande position qty=%d", pos_qty)

if signal == "buy":
if pos_qty > 0:
logging.info("Redan long. Gör inget.")
else:
# Om du råkar vara short (ovanligt i cash-konto), köp tillbaka först:
if pos_qty < 0:
submit_market_order(trading, cfg.symbol, abs(pos_qty), "buy")

qty = target_qty_for_buy(equity, last_price, cfg.max_position_pct)
submit_market_order(trading, cfg.symbol, qty, "buy")

elif signal == "sell":
if pos_qty <= 0:
logging.info("Ingen long att sälja. Gör inget.")
else:
submit_market_order(trading, cfg.symbol, pos_qty, "sell")

# Spara senaste signal (efter action)
state_set(cfg.db_path, "last_signal", signal)
# Välj kandidater och försök köpa inom dagsbudget
candidates = ranked_candidates(data_by_symbol, cfg, skills)
logging.info("Kandidater: %s", ", ".join(candidates) if candidates else "inga")
buy_candidates(trading, candidates, prices, positions, equity, cfg, cfg.db_path)


if __name__ == "__main__":
Expand Down