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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
"""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")
)
from decimal import Decimal
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
|
