Compare commits

...
2 Commits
Author SHA1 Message Date
matitos 97fbd5901e Merge pull request 'Bitcoin favicon, more strategies, default timeframe 5m' (#2) from feat/strategies-favicon-5m into main
CI/CD / backend (push) Successful in 5m23s
CI/CD / frontend (push) Successful in 5m3s
CI/CD / build (push) Successful in 16s
CI/CD / cleanup (push) Successful in 4s
CI/CD / deploy (push) Successful in 23s
2026-09-18 22:14:39 +02:00
matitos c500fb7572 Bitcoin favicon, more strategies, default timeframe 5m
CI/CD / backend (pull_request) Successful in 5m21s
CI/CD / frontend (pull_request) Successful in 5m4s
CI/CD / build (pull_request) Skipped
CI/CD / deploy (pull_request) Skipped
CI/CD / cleanup (pull_request) Skipped
- Add an SVG Bitcoin favicon and link it in index.html.
- New indicators: macd() and roc().
- Five new strategies: macd_trend, bollinger_reversion, donchian_breakout,
  ema_cross, momentum (plus the existing sma_cross, rsi_reversion).
- Default TIMEFRAME is now 5m (config, .env.example, compose, workflow, UI).
2026-09-18 20:03:30 +00:00
18 changed files with 488 additions and 11 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ DATABASE_URL=sqlite:///./data/trader.db
MODE=paper
EXCHANGE=binance
SYMBOL=BTC/USDT
TIMEFRAME=1m
TIMEFRAME=5m
# How often the engine polls for new candles, in seconds
POLL_INTERVAL=5
# Candles fetched per tick and kept for indicators
+1 -1
View File
@@ -109,7 +109,7 @@ jobs:
CRYPTO_MODE: ${{ vars.CRYPTO_MODE || 'paper' }}
CRYPTO_EXCHANGE: ${{ vars.CRYPTO_EXCHANGE || 'binance' }}
CRYPTO_SYMBOL: ${{ vars.CRYPTO_SYMBOL || 'BTC/USDT' }}
CRYPTO_TIMEFRAME: ${{ vars.CRYPTO_TIMEFRAME || '1m' }}
CRYPTO_TIMEFRAME: ${{ vars.CRYPTO_TIMEFRAME || '5m' }}
CRYPTO_POLL_INTERVAL: ${{ vars.CRYPTO_POLL_INTERVAL || '5' }}
CRYPTO_STARTING_BALANCE: ${{ vars.CRYPTO_STARTING_BALANCE || '10000' }}
CRYPTO_CORS_ORIGINS: ${{ vars.CRYPTO_CORS_ORIGINS || format('https://crypto.{0}', secrets.BASE_DOMAIN || 'matitos.org') }}
+5 -3
View File
@@ -101,7 +101,7 @@ All settings are environment variables (see `.env.example`):
| `MODE` | `paper` | `paper` or `live` |
| `EXCHANGE` | `binance` | Market data / execution venue |
| `SYMBOL` | `BTC/USDT` | Default trading pair |
| `TIMEFRAME` | `1m` | Candle interval |
| `TIMEFRAME` | `5m` | Candle interval |
| `POLL_INTERVAL` | `5` | Seconds between ticks |
| `STARTING_BALANCE` | `10000` | Paper account size |
| `FEE_BPS` / `SLIPPAGE_BPS` | `10` / `5` | Paper fill model |
@@ -159,8 +159,10 @@ from `https://api.telegram.org/bot<token>/getUpdates`.
Drop a file in `backend/app/engine/strategies/` — it is auto-discovered and
appears in the UI immediately. See **[AGENTS.md](AGENTS.md)** for the full
template and the `StrategyContext` API. Two examples ship in the repo:
`sma_cross` and `rsi_reversion`.
template and the `StrategyContext` API.
Bundled strategies: `sma_cross`, `ema_cross`, `macd_trend`,
`bollinger_reversion`, `donchian_breakout`, `momentum`, `rsi_reversion`.
## API
+1 -1
View File
@@ -27,7 +27,7 @@ class Settings(BaseSettings):
mode: str = "paper"
exchange: str = "binance"
symbol: str = "BTC/USDT"
timeframe: str = "1m"
timeframe: str = "5m"
poll_interval: float = 5.0
candle_limit: int = 200
+38
View File
@@ -113,6 +113,44 @@ def rolling_std(values: Sequence[float], period: int) -> list[Number]:
return out
def macd(
values: Sequence[float],
fast: int = 12,
slow: int = 26,
signal: int = 9,
) -> tuple[list[Number], list[Number], list[Number]]:
"""Return (macd line, signal line, histogram), each aligned with ``values``."""
ema_fast = ema(values, fast)
ema_slow = ema(values, slow)
macd_line: list[Number] = [
(f - s) if f is not None and s is not None else None
for f, s in zip(ema_fast, ema_slow, strict=True)
]
signal_line: list[Number] = [None] * len(values)
start = next((i for i, value in enumerate(macd_line) if value is not None), None)
if start is not None:
valid = [value for value in macd_line[start:] if value is not None]
for offset, value in enumerate(ema(valid, signal)):
signal_line[start + offset] = value
histogram: list[Number] = [
(m - s) if m is not None and s is not None else None
for m, s in zip(macd_line, signal_line, strict=True)
]
return macd_line, signal_line, histogram
def roc(values: Sequence[float], period: int = 10) -> list[Number]:
"""Rate of change over ``period`` bars, as a fraction (0.02 = +2%)."""
if period <= 0:
raise ValueError("period must be positive")
out: list[Number] = [None] * len(values)
for i in range(period, len(values)):
previous = values[i - period]
if previous:
out[i] = (values[i] - previous) / previous
return out
def last(series: Sequence[Number]) -> Number:
for value in reversed(series):
if value is not None:
+10 -1
View File
@@ -5,7 +5,7 @@ from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, ClassVar, Literal, Protocol
from app.engine.indicators import atr, ema, rolling_std, rsi, sma
from app.engine.indicators import atr, ema, macd, roc, rolling_std, rsi, sma
from app.models import PositionSide
from app.schemas import Candle, ParamSpec, StrategyOut
@@ -106,6 +106,15 @@ class StrategyContext:
def std(self, period: int = 20) -> float | None:
return rolling_std(self.closes, period)[-1]
def macd(
self, fast: int = 12, slow: int = 26, signal: int = 9
) -> tuple[float | None, float | None, float | None]:
macd_line, signal_line, histogram = macd(self.closes, fast, slow, signal)
return macd_line[-1], signal_line[-1], histogram[-1]
def roc(self, period: int = 10) -> float | None:
return roc(self.closes, period)[-1]
# --- order intents ------------------------------------------------
def enter_long(
self,
@@ -0,0 +1,63 @@
from __future__ import annotations
from typing import ClassVar
from app.engine.strategies.base import Strategy, StrategyContext, register
from app.models import PositionSide
@register
class BollingerReversion(Strategy):
"""Buy the lower Bollinger band, exit at the middle band."""
key: ClassVar[str] = "bollinger_reversion"
name: ClassVar[str] = "Bollinger Reversion"
description: ClassVar[str] = (
"Enter long when the close falls below the lower Bollinger band and exit "
"when it returns to the middle band (SMA). Optional ATR stop."
)
default_params: ClassVar[dict] = {
"period": 20,
"num_std": 2.0,
"atr_period": 14,
"atr_stop_mult": 0.0,
"size": 1.0,
}
param_help: ClassVar[dict] = {
"period": "Bollinger lookback",
"num_std": "Band width in standard deviations",
"atr_period": "ATR lookback for the stop",
"atr_stop_mult": "Stop distance in ATRs (0 disables)",
"size": "Fraction of equity per trade",
}
param_bounds: ClassVar[dict] = {
"period": (5, 100),
"num_std": (0.5, 4.0),
"atr_stop_mult": (0.0, 10.0),
"size": (0.01, 1.0),
}
def on_candle(self, ctx: StrategyContext) -> None:
period = int(self.params["period"])
if len(ctx.closes) < period:
return
middle = ctx.sma(period)
deviation = ctx.std(period)
if middle is None or deviation is None:
return
lower = middle - float(self.params["num_std"]) * deviation
if ctx.is_flat and ctx.price < lower:
stop = None
mult = float(self.params["atr_stop_mult"])
if mult > 0:
atr_value = ctx.atr(int(self.params["atr_period"]))
if atr_value:
stop = ctx.price - mult * atr_value
ctx.enter_long(
size=float(self.params["size"]),
stop_loss=stop,
reason="close below lower band",
)
elif ctx.side == PositionSide.LONG and ctx.price >= middle:
ctx.close(reason="close back at middle band")
@@ -0,0 +1,64 @@
from __future__ import annotations
from typing import ClassVar
from app.engine.strategies.base import Strategy, StrategyContext, register
from app.models import PositionSide
@register
class DonchianBreakout(Strategy):
"""Breakout of the highest high, exit on the lowest low."""
key: ClassVar[str] = "donchian_breakout"
name: ClassVar[str] = "Donchian Breakout"
description: ClassVar[str] = (
"Open long when the close breaks above the highest high of the entry "
"lookback; exit on a close below the lowest low of the exit lookback. "
"Optional ATR stop."
)
default_params: ClassVar[dict] = {
"entry_period": 20,
"exit_period": 10,
"atr_period": 14,
"atr_stop_mult": 2.0,
"size": 1.0,
}
param_help: ClassVar[dict] = {
"entry_period": "Breakout lookback for entries",
"exit_period": "Lookback for the exit channel",
"atr_period": "ATR lookback for the stop",
"atr_stop_mult": "Stop distance in ATRs (0 disables)",
"size": "Fraction of equity per trade",
}
param_bounds: ClassVar[dict] = {
"entry_period": (2, 200),
"exit_period": (2, 200),
"atr_stop_mult": (0.0, 10.0),
"size": (0.01, 1.0),
}
def on_candle(self, ctx: StrategyContext) -> None:
entry_period = int(self.params["entry_period"])
exit_period = int(self.params["exit_period"])
if len(ctx.candles) < max(entry_period, exit_period) + 1:
return
# Exclude the current candle from the channel.
recent_high = max(c.high for c in ctx.candles[-entry_period - 1 : -1])
recent_low = min(c.low for c in ctx.candles[-exit_period - 1 : -1])
if ctx.is_flat and ctx.price > recent_high:
stop = None
mult = float(self.params["atr_stop_mult"])
if mult > 0:
atr_value = ctx.atr(int(self.params["atr_period"]))
if atr_value:
stop = ctx.price - mult * atr_value
ctx.enter_long(
size=float(self.params["size"]),
stop_loss=stop,
reason=f"broke {entry_period}-bar high",
)
elif ctx.side == PositionSide.LONG and ctx.price < recent_low:
ctx.close(reason=f"broke {exit_period}-bar low")
@@ -0,0 +1,60 @@
from __future__ import annotations
from typing import ClassVar
from app.engine.indicators import ema, sma
from app.engine.strategies.base import Strategy, StrategyContext, register
from app.models import PositionSide
@register
class EmaCross(Strategy):
"""EMA crossover with an optional long-term trend filter."""
key: ClassVar[str] = "ema_cross"
name: ClassVar[str] = "EMA Crossover"
description: ClassVar[str] = (
"Open long when the fast EMA crosses above the slow EMA (and price is "
"above the trend filter, if enabled); exit on the opposite cross."
)
default_params: ClassVar[dict] = {
"fast": 9,
"slow": 21,
"trend_filter": 200,
"size": 1.0,
}
param_help: ClassVar[dict] = {
"fast": "Fast EMA period",
"slow": "Slow EMA period",
"trend_filter": "SMA trend filter period (0 disables)",
"size": "Fraction of equity per trade",
}
param_bounds: ClassVar[dict] = {"fast": (2, 100), "slow": (3, 300), "size": (0.01, 1.0)}
def on_candle(self, ctx: StrategyContext) -> None:
fast_period = int(self.params["fast"])
slow_period = int(self.params["slow"])
trend_period = int(self.params["trend_filter"])
needed = max(slow_period, trend_period) + 2
if len(ctx.closes) < needed:
return
fast = ema(ctx.closes, fast_period)
slow = ema(ctx.closes, slow_period)
f_prev, f_now = fast[-2], fast[-1]
s_prev, s_now = slow[-2], slow[-1]
if None in (f_prev, f_now, s_prev, s_now):
return
trend_ok = True
if trend_period > 0:
trend = sma(ctx.closes, trend_period)[-1]
trend_ok = trend is not None and ctx.price > trend
crossed_up = f_prev <= s_prev and f_now > s_now
crossed_down = f_prev >= s_prev and f_now < s_now
if ctx.is_flat and crossed_up and trend_ok:
ctx.enter_long(size=float(self.params["size"]), reason="EMA cross up")
elif ctx.side == PositionSide.LONG and crossed_down:
ctx.close(reason="EMA cross down")
@@ -0,0 +1,50 @@
from __future__ import annotations
from typing import ClassVar
from app.engine.indicators import macd
from app.engine.strategies.base import Strategy, StrategyContext, register
from app.models import PositionSide
@register
class MacdTrend(Strategy):
"""MACD crossover trend following, long-only."""
key: ClassVar[str] = "macd_trend"
name: ClassVar[str] = "MACD Trend"
description: ClassVar[str] = (
"Open long when the MACD line crosses above its signal line and close when "
"it crosses back below."
)
default_params: ClassVar[dict] = {"fast": 12, "slow": 26, "signal": 9, "size": 1.0}
param_help: ClassVar[dict] = {
"fast": "Fast EMA period",
"slow": "Slow EMA period",
"signal": "Signal EMA period",
"size": "Fraction of equity per trade",
}
param_bounds: ClassVar[dict] = {"fast": (2, 100), "slow": (3, 200), "signal": (2, 100)}
def on_candle(self, ctx: StrategyContext) -> None:
closes = ctx.closes
if len(closes) < int(self.params["slow"]) + int(self.params["signal"]) + 2:
return
macd_line, signal_line, _ = macd(
closes,
int(self.params["fast"]),
int(self.params["slow"]),
int(self.params["signal"]),
)
m_prev, m_now = macd_line[-2], macd_line[-1]
s_prev, s_now = signal_line[-2], signal_line[-1]
if None in (m_prev, m_now, s_prev, s_now):
return
crossed_up = m_prev <= s_prev and m_now > s_now
crossed_down = m_prev >= s_prev and m_now < s_now
if ctx.is_flat and crossed_up:
ctx.enter_long(size=float(self.params["size"]), reason="MACD crossed above signal")
elif ctx.side == PositionSide.LONG and crossed_down:
ctx.close(reason="MACD crossed below signal")
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
from typing import ClassVar
from app.engine.strategies.base import Strategy, StrategyContext, register
from app.models import PositionSide
@register
class Momentum(Strategy):
"""Rate-of-change momentum, long-only."""
key: ClassVar[str] = "momentum"
name: ClassVar[str] = "Momentum (ROC)"
description: ClassVar[str] = (
"Open long when the rate of change over the lookback exceeds the entry "
"threshold; exit when momentum turns negative."
)
default_params: ClassVar[dict] = {
"period": 10,
"entry_threshold": 0.01,
"atr_period": 14,
"atr_stop_mult": 2.0,
"size": 1.0,
}
param_help: ClassVar[dict] = {
"period": "ROC lookback in bars",
"entry_threshold": "Minimum ROC to enter (0.01 = 1%)",
"atr_period": "ATR lookback for the stop",
"atr_stop_mult": "Stop distance in ATRs (0 disables)",
"size": "Fraction of equity per trade",
}
param_bounds: ClassVar[dict] = {
"period": (2, 200),
"entry_threshold": (0.0, 1.0),
"atr_stop_mult": (0.0, 10.0),
"size": (0.01, 1.0),
}
def on_candle(self, ctx: StrategyContext) -> None:
value = ctx.roc(int(self.params["period"]))
if value is None:
return
if ctx.is_flat and value > float(self.params["entry_threshold"]):
stop = None
mult = float(self.params["atr_stop_mult"])
if mult > 0:
atr_value = ctx.atr(int(self.params["atr_period"]))
if atr_value:
stop = ctx.price - mult * atr_value
ctx.enter_long(
size=float(self.params["size"]),
stop_loss=stop,
reason=f"ROC {value:.2%} above threshold",
)
elif ctx.side == PositionSide.LONG and value < 0:
ctx.close(reason=f"ROC {value:.2%} turned negative")
+23 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
import pytest
from app.engine.indicators import atr, ema, rolling_std, rsi, sma
from app.engine.indicators import atr, ema, macd, roc, rolling_std, rsi, sma
from app.schemas import Candle
@@ -38,3 +38,25 @@ def test_atr_flat_market_is_zero() -> None:
def test_rolling_std() -> None:
result = rolling_std([1, 1, 1, 1], 2)
assert result[-1] == pytest.approx(0.0)
def test_roc() -> None:
values = [100.0, 110.0, 121.0, 100.0]
out = roc(values, 2)
assert out[0] is None and out[1] is None
assert out[2] == pytest.approx(0.21)
assert out[3] == pytest.approx(-0.09090909, rel=1e-4)
def test_macd_shapes_and_crossover_sign() -> None:
rising = [float(i) for i in range(1, 80)]
macd_line, signal_line, histogram = macd(rising, 12, 26, 9)
assert len(macd_line) == len(rising)
assert macd_line[-1] is not None and signal_line[-1] is not None
# In a steady uptrend MACD is positive (but can be below its signal after lag).
assert macd_line[-1] > 0
assert histogram[-1] == pytest.approx(macd_line[-1] - signal_line[-1])
falling = [float(i) for i in range(80, 0, -1)]
macd_falling, _, _ = macd(falling, 12, 26, 9)
assert macd_falling[-1] < 0
+99
View File
@@ -0,0 +1,99 @@
from __future__ import annotations
from collections.abc import Sequence
from app.engine.strategies.base import StrategyContext
from app.schemas import Candle
def make_candles(closes: Sequence[float]) -> list[Candle]:
return [Candle(ts=i, open=c, high=c, low=c, close=c, volume=1.0) for i, c in enumerate(closes)]
def context(closes: Sequence[float], position=None) -> StrategyContext:
candles = make_candles(closes)
return StrategyContext(candles=candles, position=position, equity=1000.0, cash=1000.0)
def test_all_strategies_discovered() -> None:
from app.engine.strategies import list_strategies
keys = {cls.key for cls in list_strategies()}
assert {
"sma_cross",
"rsi_reversion",
"macd_trend",
"bollinger_reversion",
"donchian_breakout",
"ema_cross",
"momentum",
} <= keys
def test_every_strategy_has_spec_and_params() -> None:
from app.engine.strategies import list_strategies, strategy_spec
for cls in list_strategies():
spec = strategy_spec(cls)
assert spec.key and spec.name and spec.description
# Defaults must round-trip through the constructor and be constructible.
instance = cls({})
assert instance.key == cls.key
# A short synthetic series must never raise for a registered strategy.
ctx = context([100, 101, 100, 102, 99, 103, 98, 104, 97, 105])
instance.on_candle(ctx)
def test_ema_cross_enters_on_cross_up() -> None:
from app.engine.strategies import get_strategy
strategy = get_strategy("ema_cross")({"fast": 2, "slow": 4, "trend_filter": 0, "size": 0.5})
# Fast EMA dips below the slow EMA, then crosses back above.
ctx = context([10, 10, 10, 10, 10, 10, 10, 10, 7, 9, 12])
strategy.on_candle(ctx)
intents = ctx.take_intents()
assert intents and intents[0].action == "enter_long"
def test_momentum_enters_on_positive_roc() -> None:
from app.engine.strategies import get_strategy
strategy = get_strategy("momentum")({"period": 3, "entry_threshold": 0.01, "atr_stop_mult": 0})
ctx = context([100, 100, 100, 105])
strategy.on_candle(ctx)
intents = ctx.take_intents()
assert intents and intents[0].action == "enter_long"
def test_macd_trend_no_intent_when_flat_short_history() -> None:
from app.engine.strategies import get_strategy
strategy = get_strategy("macd_trend")({})
ctx = context([100, 101, 102])
strategy.on_candle(ctx)
assert ctx.take_intents() == []
def test_bollinger_enters_below_lower_band() -> None:
from app.engine.strategies import get_strategy
strategy = get_strategy("bollinger_reversion")(
{"period": 5, "num_std": 1.0, "atr_stop_mult": 0}
)
# Flat then a sharp drop below the lower band.
ctx = context([100, 100, 100, 100, 80])
strategy.on_candle(ctx)
intents = ctx.take_intents()
assert intents and intents[0].action == "enter_long"
def test_donchian_enters_on_breakout() -> None:
from app.engine.strategies import get_strategy
strategy = get_strategy("donchian_breakout")(
{"entry_period": 5, "exit_period": 3, "atr_stop_mult": 0}
)
ctx = context([100, 100, 100, 100, 100, 110])
strategy.on_candle(ctx)
intents = ctx.take_intents()
assert intents and intents[0].action == "enter_long"
+1 -1
View File
@@ -12,7 +12,7 @@ services:
- MODE=${MODE:-paper}
- EXCHANGE=${EXCHANGE:-binance}
- SYMBOL=${SYMBOL:-BTC/USDT}
- TIMEFRAME=${TIMEFRAME:-1m}
- TIMEFRAME=${TIMEFRAME:-5m}
- POLL_INTERVAL=${POLL_INTERVAL:-5}
- STARTING_BALANCE=${STARTING_BALANCE:-10000}
- FEE_BPS=${FEE_BPS:-10}
+1 -1
View File
@@ -22,7 +22,7 @@ services:
- MODE=${CRYPTO_MODE:-paper}
- EXCHANGE=${CRYPTO_EXCHANGE:-binance}
- SYMBOL=${CRYPTO_SYMBOL:-BTC/USDT}
- TIMEFRAME=${CRYPTO_TIMEFRAME:-1m}
- TIMEFRAME=${CRYPTO_TIMEFRAME:-5m}
- POLL_INTERVAL=${CRYPTO_POLL_INTERVAL:-5}
- STARTING_BALANCE=${CRYPTO_STARTING_BALANCE:-10000}
- FEE_BPS=${CRYPTO_FEE_BPS:-10}
+3
View File
@@ -3,6 +3,9 @@
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<link rel="icon" type="image/svg+xml" href="/favicon.svg" />
<link rel="apple-touch-icon" href="/favicon.svg" />
<meta name="theme-color" content="#F7931A" />
<title>Crypto Trader</title>
</head>
<body class="bg-surface text-slate-100">
+9
View File
@@ -0,0 +1,9 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 32 32" width="32" height="32">
<circle cx="16" cy="16" r="16" fill="#F7931A"/>
<text x="16" y="23.5" text-anchor="middle" font-family="Arial, Helvetica, sans-serif"
font-size="21" font-weight="bold" fill="#ffffff">B</text>
<rect x="13.9" y="4.2" width="2.2" height="4.6" rx="1.1" fill="#ffffff"/>
<rect x="17.6" y="4.2" width="2.2" height="4.6" rx="1.1" fill="#ffffff"/>
<rect x="13.9" y="23.2" width="2.2" height="4.6" rx="1.1" fill="#ffffff"/>
<rect x="17.6" y="23.2" width="2.2" height="4.6" rx="1.1" fill="#ffffff"/>
</svg>

After

Width:  |  Height:  |  Size: 601 B

+1 -1
View File
@@ -8,7 +8,7 @@ const props = defineProps({
submitLabel: { type: String, default: 'Run' },
submitClass: { type: String, default: 'btn-primary' },
symbol: { type: String, default: 'BTC/USDT' },
timeframe: { type: String, default: '1m' },
timeframe: { type: String, default: '5m' },
presets: { type: Array, default: () => [] },
})
const emit = defineEmits(['submit', 'save-preset', 'delete-preset', 'load-preset'])