-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: SA-421 Add SDK support for RollingBondStrategy & SingleBondStra…
…tegy
- Loading branch information
1 parent
1005ab3
commit 94d5260
Showing
3 changed files
with
141 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,36 @@ | ||
import logging | ||
|
||
import sigtech.api as sig | ||
|
||
logging.basicConfig(level=logging.INFO) | ||
env = sig.init() | ||
|
||
|
||
# Create a strategy to buy a bond and reinvest coupons | ||
single_bond = sig.SingleBondStrategy( | ||
currency="USD", | ||
bond_name="US 2.25 2046/08/15 GOVT", # US912810RT79 | ||
) | ||
print(single_bond.history()) | ||
print(single_bond.currency) | ||
print(single_bond.start_date) | ||
|
||
|
||
# Create a strategy to regularly buy bonds and reinvest coupons | ||
rolling_bond_strategy = sig.RollingBondStrategy( | ||
country="US", | ||
tenor="10Y", | ||
start_date="2016-08-15", | ||
) | ||
print(rolling_bond_strategy.history()) | ||
print(rolling_bond_strategy.currency) | ||
print(rolling_bond_strategy.tenor) | ||
print(rolling_bond_strategy.country) | ||
print(rolling_bond_strategy.start_date) | ||
|
||
df_portfolio_timeline = rolling_bond_strategy.plot.portfolio_table( | ||
dts="ACTION_PTS", | ||
unit_type="TRADE", | ||
as_df=True, | ||
) | ||
print(df_portfolio_timeline) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,99 @@ | ||
import datetime as dtm | ||
from typing import Optional, Union | ||
|
||
from sigtech.api.client.response import Response | ||
from sigtech.api.framework.environment import env | ||
from sigtech.api.framework.strategies.strategy import Strategy | ||
|
||
|
||
class SingleBondStrategy(Strategy): | ||
""" | ||
Strategy assigning weight to a single bond instrument and reinvesting bond coupons. | ||
Keyword arguments: | ||
* ``bond_name``: Name of bond to trade. | ||
""" | ||
|
||
def __init__( | ||
self, | ||
bond_name: str, | ||
currency: Optional[str] = None, # QUANT-1537 | ||
): | ||
super().__init__( | ||
currency=currency, | ||
bond_name=bond_name, | ||
) | ||
|
||
def _get_strategy_obj(self, session_id: str, **inputs) -> Response: | ||
""" | ||
Fetch rolling bond strategy from API. | ||
""" | ||
api_inputs = {k: v for k, v in inputs.items() if v is not None} | ||
api_inputs["identifier"] = api_inputs.pop("bond_name") | ||
del api_inputs["currency"] | ||
return env().client.strategies.bonds.reinvestment.create( | ||
session_id=session_id, | ||
**api_inputs, | ||
) | ||
|
||
@property | ||
def currency(self) -> str: | ||
return self._get_reference_data()["currency"] | ||
|
||
@property | ||
def start_date(self) -> str: | ||
return self._get_reference_data()["startDate"] | ||
|
||
|
||
class RollingBondStrategy(Strategy): | ||
""" | ||
Rolling bond strategy that rolls bonds. | ||
Invests in the underlying bonds using SingleBondStrategy, coupons are reinvested. | ||
Keyword arguments: | ||
* ``country``: Two-letter country code, e.g. ``'US'`` or ``'GB'``. | ||
* ``tenor``: Tenor string in ``'[xx]Y'`` format, e.g. ``'10Y'`` or ``'5Y'``. | ||
* ``start_date``: Start date of the strategy. | ||
""" | ||
|
||
def __init__( | ||
self, | ||
country: str, | ||
tenor: str, | ||
start_date: Optional[Union[str, dtm.date]] = None, | ||
): | ||
start_date = str(start_date) if isinstance(start_date, dtm.date) else start_date | ||
super().__init__( | ||
country=country, | ||
tenor=tenor, | ||
start_date=start_date, | ||
) | ||
|
||
def _get_strategy_obj(self, session_id: str, **inputs) -> Response: | ||
""" | ||
Fetch rolling bond strategy from API. | ||
""" | ||
api_inputs = {k: v for k, v in inputs.items() if v is not None} | ||
return env().client.strategies.bonds.rolling.create( | ||
session_id=session_id, | ||
**api_inputs, | ||
) | ||
|
||
@property | ||
def currency(self) -> str: | ||
return self._get_reference_data()["currency"] | ||
|
||
@property | ||
def tenor(self) -> str: | ||
return self._get_reference_data()["tenor"] | ||
|
||
@property | ||
def country(self) -> str: | ||
return self._get_reference_data()["country"] | ||
|
||
@property | ||
def start_date(self) -> str: | ||
return self._get_reference_data()["startDate"] |