matitos
9c3bee87ac
FastAPI backend with a single-strategy async engine, paper and Binance brokers, pure-python indicators, a backtester that shares the engine's portfolio math, and a Vue dashboard with a live WebSocket feed. Adds the Gitea Actions CI/CD workflow that deploys to Traefik via docker_default.
69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from app.engine.strategies import get_strategy, list_strategies, strategy_spec
|
|
from app.engine.strategies.base import StrategyContext
|
|
from app.models import Position, PositionSide
|
|
from app.schemas import Candle
|
|
|
|
|
|
def make_candles(closes: list[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 test_examples_are_registered() -> None:
|
|
keys = {cls.key for cls in list_strategies()}
|
|
assert {"sma_cross", "rsi_reversion"} <= keys
|
|
|
|
|
|
def test_param_specs_expose_defaults() -> None:
|
|
spec = strategy_spec(get_strategy("sma_cross"))
|
|
names = {param.name for param in spec.params}
|
|
assert {"fast", "slow", "size"} == names
|
|
fast = next(param for param in spec.params if param.name == "fast")
|
|
assert fast.default == 10
|
|
assert fast.type == "int"
|
|
|
|
|
|
def test_unknown_params_rejected() -> None:
|
|
cls = get_strategy("sma_cross")
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError):
|
|
cls({"nope": 1})
|
|
|
|
|
|
def test_sma_cross_emits_enter_long() -> None:
|
|
strategy = get_strategy("sma_cross")({"fast": 2, "slow": 4, "size": 0.5})
|
|
ctx = StrategyContext(
|
|
candles=make_candles([10, 10, 10, 10, 10, 11]),
|
|
position=None,
|
|
equity=1000,
|
|
cash=1000,
|
|
)
|
|
strategy.on_candle(ctx)
|
|
intents = ctx.take_intents()
|
|
assert len(intents) == 1
|
|
assert intents[0].action == "enter_long"
|
|
assert intents[0].size == 0.5
|
|
|
|
|
|
def test_sma_cross_emits_close_on_cross_down() -> None:
|
|
strategy = get_strategy("sma_cross")({"fast": 2, "slow": 4, "size": 0.5})
|
|
position = Position(
|
|
strategy_key="sma_cross",
|
|
symbol="BTC/USDT",
|
|
side=PositionSide.LONG,
|
|
qty=1.0,
|
|
entry_price=10.0,
|
|
)
|
|
ctx = StrategyContext(
|
|
candles=make_candles([10, 10, 10, 10, 10, 11, 8]),
|
|
position=position,
|
|
equity=1000,
|
|
cash=1000,
|
|
)
|
|
strategy.on_candle(ctx)
|
|
intents = ctx.take_intents()
|
|
assert len(intents) == 1
|
|
assert intents[0].action == "close"
|