1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
"""Integration test: order -> portfolio tracker -> position state."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "services" / "portfolio-manager" / "src"))
import pytest
from decimal import Decimal
from datetime import datetime, timezone
from shared.models import Order, OrderSide, OrderType, OrderStatus
from portfolio_manager.portfolio import PortfolioTracker
def test_portfolio_tracks_buy_sell_cycle():
"""Buy then sell should update position and reset on full sell."""
tracker = PortfolioTracker()
buy_order = Order(
signal_id="sig-1", symbol="BTCUSDT", side=OrderSide.BUY,
type=OrderType.MARKET, price=Decimal("50000"),
quantity=Decimal("0.1"), status=OrderStatus.FILLED,
)
tracker.apply_order(buy_order)
pos = tracker.get_position("BTCUSDT")
assert pos is not None
assert pos.quantity == Decimal("0.1")
assert pos.avg_entry_price == Decimal("50000")
sell_order = Order(
signal_id="sig-2", symbol="BTCUSDT", side=OrderSide.SELL,
type=OrderType.MARKET, price=Decimal("55000"),
quantity=Decimal("0.1"), status=OrderStatus.FILLED,
)
tracker.apply_order(sell_order)
pos = tracker.get_position("BTCUSDT")
assert pos is None # Fully sold
def test_portfolio_weighted_average_on_multiple_buys():
"""Multiple buys at different prices should compute weighted average."""
tracker = PortfolioTracker()
tracker.apply_order(Order(
signal_id="s1", symbol="BTCUSDT", side=OrderSide.BUY,
type=OrderType.MARKET, price=Decimal("50000"),
quantity=Decimal("0.1"), status=OrderStatus.FILLED,
))
tracker.apply_order(Order(
signal_id="s2", symbol="BTCUSDT", side=OrderSide.BUY,
type=OrderType.MARKET, price=Decimal("60000"),
quantity=Decimal("0.1"), status=OrderStatus.FILLED,
))
pos = tracker.get_position("BTCUSDT")
assert pos.quantity == Decimal("0.2")
assert pos.avg_entry_price == Decimal("55000") # weighted avg
|
