Adriano
b8c2bba320
C2: fifo_roundtrips ora apre un lotto SHORT quando un sell non trova lotti long da chiudere (prima il resto spariva in silenzio), chiude prima il lato opposto e apre l'eccedenza nel segno del fill — anche a meta' dello stesso fill (inversione long<->short). stato_aperto e report() portano la direzione del lotto residuo. C10: un fill "NON VERIFICATO" a qty 0 (place_order error) non entra piu' nel FIFO ne' in stato_aperto (WHERE qty > 0 in rebuild_roundtrips/stato_aperto); resta nel DB marcato (verified=0), non e' scomparso. test_fifo_non_mescola_gli_asset aggiornato: una vendita ETH senza lotti long ora apre uno short invece di sparire (era il comportamento che C2 corregge). Test nuovi in tests/test_tradesdb_u4.py. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
121 lines
5.6 KiB
Python
121 lines
5.6 KiB
Python
"""Test nuovi per l'unita' U4 (issue #32 C2, C10) — tradesdb.py.
|
|
|
|
Non ripete la suite esistente (`test_tradesdb.py`): quella resta bit-exact sui casi
|
|
long-aperto/long-chiuso (vedi il commit). Qui solo cio' che il fix aggiunge: lotti SHORT,
|
|
inversione in un fill, e i fill a qty<=0 fuori dal FIFO.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.live import tradesdb as T # noqa: E402
|
|
|
|
|
|
def _f(ts, asset, side, qty, price, fee=0.0, verified=1):
|
|
return T.Fill(ts_utc=ts, asset=asset, side=side, qty=qty, price=price, fee=fee,
|
|
verified=verified)
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
con = T.connect(tmp_path / "t.db")
|
|
yield con
|
|
con.close()
|
|
|
|
|
|
# ---------------------------------------------------------------- short: apertura e chiusura
|
|
|
|
def test_una_vendita_senza_lotti_long_apre_un_short(db):
|
|
fills = [_f("2026-01-01T00:00:00+00:00", "BTC", "sell", 1.0, 100.0, fee=1.0)]
|
|
rts, aperti = T.fifo_roundtrips(fills)
|
|
assert rts == [] # niente da chiudere: solo apertura
|
|
assert aperti["BTC"][0]["qty"] == pytest.approx(1.0)
|
|
assert aperti["BTC"][0]["price"] == pytest.approx(100.0)
|
|
assert aperti["BTC"][0]["direzione"] == "short"
|
|
|
|
|
|
def test_un_acquisto_chiude_il_short_pnl_entry_meno_exit(db):
|
|
fills = [_f("2026-01-01T00:00:00+00:00", "BTC", "sell", 1.0, 100.0, fee=1.0), # apre short @100
|
|
_f("2026-01-02T00:00:00+00:00", "BTC", "buy", 1.0, 90.0, fee=1.0)] # chiude @90
|
|
rts, aperti = T.fifo_roundtrips(fills)
|
|
assert len(rts) == 1
|
|
r = rts[0]
|
|
assert r["direzione"] == "short"
|
|
assert r["px_in"] == 100.0 and r["px_out"] == 90.0
|
|
assert r["pnl_lordo"] == pytest.approx(10.0) # venduto alto, comprato basso: profitto
|
|
assert r["fee_quota"] == pytest.approx(2.0)
|
|
assert r["pnl_netto"] == pytest.approx(8.0)
|
|
assert aperti == {} # flat: nessun lotto residuo
|
|
|
|
|
|
def test_chiusura_parziale_di_uno_short_lascia_il_resto_short(db):
|
|
fills = [_f("2026-01-01T00:00:00+00:00", "ETH", "sell", 1.0, 100.0),
|
|
_f("2026-01-02T00:00:00+00:00", "ETH", "buy", 0.25, 80.0)]
|
|
rts, aperti = T.fifo_roundtrips(fills)
|
|
assert len(rts) == 1 and rts[0]["qty"] == pytest.approx(0.25)
|
|
assert rts[0]["pnl_lordo"] == pytest.approx(0.25 * (100.0 - 80.0))
|
|
assert aperti["ETH"][0]["qty"] == pytest.approx(0.75)
|
|
assert aperti["ETH"][0]["direzione"] == "short"
|
|
|
|
|
|
# ---------------------------------------------------------------- inversione in un solo fill
|
|
|
|
def test_inversione_long_a_short_in_un_fill_non_scarta_il_resto(db):
|
|
fills = [_f("2026-01-01T00:00:00+00:00", "BTC", "buy", 1.0, 100.0), # long 1.0 @100
|
|
_f("2026-01-02T00:00:00+00:00", "BTC", "sell", 2.0, 150.0)] # chiude + apre short
|
|
rts, aperti = T.fifo_roundtrips(fills)
|
|
assert len(rts) == 1
|
|
assert rts[0]["direzione"] == "long" and rts[0]["qty"] == pytest.approx(1.0)
|
|
assert rts[0]["pnl_lordo"] == pytest.approx(50.0) # (150-100)*1
|
|
assert aperti["BTC"][0]["qty"] == pytest.approx(1.0) # il resto: 2.0-1.0
|
|
assert aperti["BTC"][0]["price"] == pytest.approx(150.0)
|
|
assert aperti["BTC"][0]["direzione"] == "short" # nessun resto scartato
|
|
|
|
|
|
def test_inversione_short_a_long_in_un_fill(db):
|
|
fills = [_f("2026-01-01T00:00:00+00:00", "ETH", "sell", 1.0, 100.0), # short 1.0 @100
|
|
_f("2026-01-02T00:00:00+00:00", "ETH", "buy", 3.0, 90.0)] # chiude + apre long
|
|
rts, aperti = T.fifo_roundtrips(fills)
|
|
assert len(rts) == 1
|
|
assert rts[0]["direzione"] == "short" and rts[0]["qty"] == pytest.approx(1.0)
|
|
assert rts[0]["pnl_lordo"] == pytest.approx(10.0) # (100-90)*1
|
|
assert aperti["ETH"][0]["qty"] == pytest.approx(2.0) # il resto: 3.0-1.0
|
|
assert aperti["ETH"][0]["direzione"] == "long"
|
|
|
|
|
|
def test_stato_aperto_riporta_la_direzione_short(db):
|
|
T.upsert_fills(db, [_f("2026-01-01T00:00:00+00:00", "ETH", "sell", 1.0, 100.0)])
|
|
st = T.stato_aperto(db)
|
|
assert st["ETH"]["qty"] == pytest.approx(1.0)
|
|
assert st["ETH"]["direzione"] == "short"
|
|
|
|
|
|
# ---------------------------------------------------------------- #32 C10: fill a qty<=0
|
|
|
|
def test_fill_a_qty_zero_non_entra_nel_fifo(db):
|
|
T.upsert_fills(db, [
|
|
_f("2026-01-01T00:00:00+00:00", "BTC", "buy", 0.0, 100.0, verified=0), # place_order error
|
|
_f("2026-01-02T00:00:00+00:00", "BTC", "buy", 1.0, 100.0),
|
|
_f("2026-01-03T00:00:00+00:00", "BTC", "sell", 1.0, 110.0)])
|
|
assert T.rebuild_roundtrips(db) == 1 # solo il round-trip vero, non uno spurio
|
|
rt = db.execute("SELECT * FROM roundtrips").fetchall()
|
|
assert len(rt) == 1 and rt[0]["qty"] == pytest.approx(1.0)
|
|
|
|
|
|
def test_fill_a_qty_zero_non_entra_in_stato_aperto_ma_resta_nel_db(db):
|
|
T.upsert_fills(db, [
|
|
_f("2026-01-01T00:00:00+00:00", "ETH", "sell", 0.0, 100.0, verified=0), # place_order error
|
|
_f("2026-01-02T00:00:00+00:00", "ETH", "buy", 1.0, 100.0)])
|
|
st = T.stato_aperto(db)
|
|
assert st["ETH"]["qty"] == pytest.approx(1.0) # il fill a qty 0 non conta come lotto
|
|
# ma resta leggibile nel DB, marcato NON verificato — non e' sparito, e' escluso dal FIFO
|
|
righe = db.execute("SELECT qty, verified FROM fills WHERE qty = 0").fetchall()
|
|
assert len(righe) == 1 and righe[0]["verified"] == 0
|
|
assert db.execute("SELECT COUNT(*) c FROM fills").fetchone()["c"] == 2
|