Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Order(SENTINEL_DT, 'EQ:ABC', 123),
Order(SENTINEL_DT, 'EQ:DEF', 456),
Order(SENTINEL_DT, 'EQ:GHI', -217),
Order(SENTINEL_DT, 'EQ:JKL', -48)
]
),
(
'non-empty portfolios, partially-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:DEF': {'quantity': 217}, 'EQ:GHI': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:ABC', 123),
Order(SENTINEL_DT, 'EQ:DEF', 239),
Order(SENTINEL_DT, 'EQ:GHI', -48)
]
),
(
Order(SENTINEL_DT, 'EQ:ABC', 100),
Order(SENTINEL_DT, 'EQ:DEF', 250)
]
),
(
'empty target portfolio with non-empty current portfolio',
{},
{'EQ:ABC': {'quantity': 345}, 'EQ:DEF': {'quantity': 223}},
[
Order(SENTINEL_DT, 'EQ:ABC', -345),
Order(SENTINEL_DT, 'EQ:DEF', -250)
]
),
(
'non-empty portfolios, non-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:GHI': {'quantity': 217}, 'EQ:JKL': {'quantity': 48}},
Order(SENTINEL_DT, 'EQ:DEF', 408)
]
)
]
)
def test_generate_rebalance_orders(
helpers, description, target_portfolio, current_portfolio, expected
):
"""
Tests the _generate_rebalance_orders method of the
PortfolioConstructionModel base class.
"""
port_id = '1234'
broker = Mock()
universe = Mock()
order_sizer = Mock()
Order(SENTINEL_DT, 'EQ:DEF', 456),
Order(SENTINEL_DT, 'EQ:GHI', -217),
Order(SENTINEL_DT, 'EQ:JKL', -48)
]
),
(
'non-empty portfolios, partially-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:DEF': {'quantity': 217}, 'EQ:GHI': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:ABC', 123),
Order(SENTINEL_DT, 'EQ:DEF', 239),
Order(SENTINEL_DT, 'EQ:GHI', -48)
]
),
(
'non-empty portfolios, fully-intersecting symbols',
Order(SENTINEL_DT, 'EQ:DEF', 239),
Order(SENTINEL_DT, 'EQ:GHI', -48)
]
),
(
'non-empty portfolios, fully-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:ABC': {'quantity': 217}, 'EQ:DEF': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:ABC', -94),
Order(SENTINEL_DT, 'EQ:DEF', 408)
]
)
]
)
def test_generate_rebalance_orders(
helpers, description, target_portfolio, current_portfolio, expected
Order(SENTINEL_DT, 'EQ:DEF', 250)
]
),
(
'empty target portfolio with non-empty current portfolio',
{},
{'EQ:ABC': {'quantity': 345}, 'EQ:DEF': {'quantity': 223}},
[
Order(SENTINEL_DT, 'EQ:ABC', -345),
Order(SENTINEL_DT, 'EQ:DEF', -250)
]
),
(
'non-empty portfolios, non-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:GHI': {'quantity': 217}, 'EQ:JKL': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:DEF', -250)
]
),
(
'non-empty portfolios, non-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:GHI': {'quantity': 217}, 'EQ:JKL': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:ABC', 123),
Order(SENTINEL_DT, 'EQ:DEF', 456),
Order(SENTINEL_DT, 'EQ:GHI', -217),
Order(SENTINEL_DT, 'EQ:JKL', -48)
]
),
(
'non-empty portfolios, partially-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
Order(SENTINEL_DT, 'EQ:JKL', -48)
]
),
(
'non-empty portfolios, partially-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:DEF': {'quantity': 217}, 'EQ:GHI': {'quantity': 48}},
[
Order(SENTINEL_DT, 'EQ:ABC', 123),
Order(SENTINEL_DT, 'EQ:DEF', 239),
Order(SENTINEL_DT, 'EQ:GHI', -48)
]
),
(
'non-empty portfolios, fully-intersecting symbols',
{'EQ:ABC': {'quantity': 123}, 'EQ:DEF': {'quantity': 456}},
{'EQ:ABC': {'quantity': 217}, 'EQ:DEF': {'quantity': 48}},
if asset not in target_portfolio:
target_portfolio[asset] = {"quantity": 0}
# Iterate through the asset list and create the difference
# quantities required for each asset
rebalance_portfolio = {}
for asset in target_portfolio.keys():
target_qty = target_portfolio[asset]["quantity"]
current_qty = current_portfolio[asset]["quantity"]
order_qty = target_qty - current_qty
rebalance_portfolio[asset] = {"quantity": order_qty}
# Create the rebalancing Order list from the order portfolio
# only where quantities are non-zero
rebalance_orders = [
Order(dt, asset, rebalance_portfolio[asset]["quantity"])
for asset, asset_dict in sorted(
rebalance_portfolio.items(), key=lambda x: x[0]
)
if rebalance_portfolio[asset]["quantity"] != 0
]
return rebalance_orders