Compare commits
4 Commits
faf0b44e46
...
fa2ab84c92
Author | SHA1 | Date | |
---|---|---|---|
fa2ab84c92 | |||
6ffa52f84e | |||
d88acc5888 | |||
eb63766c1e |
@ -21,14 +21,14 @@ Fincal aims to simplify things by allowing you to:
|
|||||||
### Core features
|
### Core features
|
||||||
- [ ] Add __setitem__
|
- [ ] Add __setitem__
|
||||||
- [ ] Create emtpy TimeSeries object
|
- [ ] Create emtpy TimeSeries object
|
||||||
- [ ] Read from CSV
|
- [x] Read from CSV
|
||||||
- [ ] Write to CSV
|
- [ ] Write to CSV
|
||||||
- [ ] Convert to dict
|
- [ ] Convert to dict
|
||||||
- [ ] Convert to list of dicts
|
- [ ] Convert to list of dicts
|
||||||
### Fincal features
|
### Fincal features
|
||||||
- [ ] Sync two TimeSeries
|
- [ ] Sync two TimeSeries
|
||||||
- [ ] Average rolling return
|
- [x] Average rolling return
|
||||||
- [ ] Sharpe ratio
|
- [ ] Sharpe ratio
|
||||||
- [ ] Jensen's Alpha
|
- [ ] Jensen's Alpha
|
||||||
- [ ] Beta
|
- [ ] Beta
|
||||||
- [ ] Max drawdown
|
- [x] Max drawdown
|
@ -1,9 +1,11 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import csv
|
||||||
import datetime
|
import datetime
|
||||||
import math
|
import math
|
||||||
|
import pathlib
|
||||||
import statistics
|
import statistics
|
||||||
from typing import Iterable, List, Literal, Mapping, TypedDict, Union
|
from typing import Iterable, List, Literal, Mapping, Tuple, TypedDict, Union
|
||||||
|
|
||||||
from dateutil.relativedelta import relativedelta
|
from dateutil.relativedelta import relativedelta
|
||||||
|
|
||||||
@ -581,6 +583,65 @@ class TimeSeries(TimeSeriesCore):
|
|||||||
return output_ts
|
return output_ts
|
||||||
|
|
||||||
|
|
||||||
|
def _preprocess_csv(file_path: str | pathlib.Path, delimiter: str = ",", encoding: str = "utf-8") -> List[list]:
|
||||||
|
"""Preprocess csv data"""
|
||||||
|
|
||||||
|
if isinstance(file_path, str):
|
||||||
|
file_path = pathlib.Path(file_path)
|
||||||
|
|
||||||
|
if not file_path.exists():
|
||||||
|
raise ValueError("File not found. Check the file path")
|
||||||
|
|
||||||
|
with open(file_path, "r", encoding=encoding) as file:
|
||||||
|
reader = csv.reader(file, delimiter=delimiter)
|
||||||
|
csv_data = list(reader)
|
||||||
|
|
||||||
|
csv_data = [i for i in csv_data if i] # remove blank rows
|
||||||
|
if not csv_data:
|
||||||
|
raise ValueError("File is empty")
|
||||||
|
|
||||||
|
return csv_data
|
||||||
|
|
||||||
|
|
||||||
|
def read_csv(
|
||||||
|
csv_file_path: str | pathlib.Path,
|
||||||
|
frequency: Literal["D", "W", "M", "Q", "Y"],
|
||||||
|
date_format: str = None,
|
||||||
|
col_names: Tuple[str, str] = None,
|
||||||
|
col_index: Tuple[int, int] = (0, 1),
|
||||||
|
has_header: bool = True,
|
||||||
|
skip_rows: int = 0,
|
||||||
|
nrows: int = -1,
|
||||||
|
delimiter: str = ",",
|
||||||
|
encoding: str = "utf-8",
|
||||||
|
) -> TimeSeriesCore:
|
||||||
|
"""Reads Time Series data directly from a CSV file"""
|
||||||
|
|
||||||
|
data = _preprocess_csv(csv_file_path, delimiter, encoding)
|
||||||
|
|
||||||
|
read_start_row = skip_rows
|
||||||
|
read_end_row = skip_rows + nrows if nrows >= 0 else None
|
||||||
|
|
||||||
|
if has_header:
|
||||||
|
header = data[read_start_row]
|
||||||
|
print(header)
|
||||||
|
# fmt: off
|
||||||
|
# Black and pylance disagree on the foratting of the following line, hence formatting is disabled
|
||||||
|
data = data[(read_start_row + 1):read_end_row]
|
||||||
|
# fmt: on
|
||||||
|
|
||||||
|
if col_names is not None:
|
||||||
|
date_col = header.index(col_names[0])
|
||||||
|
value_col = header.index(col_names[1])
|
||||||
|
else:
|
||||||
|
date_col = col_index[0]
|
||||||
|
value_col = col_index[1]
|
||||||
|
|
||||||
|
ts_data = [(i[date_col], i[value_col]) for i in data if i]
|
||||||
|
|
||||||
|
return TimeSeries(ts_data, frequency=frequency, date_format=date_format)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
date_series = [
|
date_series = [
|
||||||
datetime.datetime(2020, 1, 11),
|
datetime.datetime(2020, 1, 11),
|
||||||
|
@ -116,7 +116,7 @@ def _find_closest_date(
|
|||||||
raise ValueError(f"Invalid argument for if_not_found: {if_not_found}")
|
raise ValueError(f"Invalid argument for if_not_found: {if_not_found}")
|
||||||
|
|
||||||
|
|
||||||
def _interval_to_years(interval_type: Literal["years", "months", "day"], interval_value: int) -> int:
|
def _interval_to_years(interval_type: Literal["years", "months", "day"], interval_value: int) -> float:
|
||||||
"""Converts any time period to years for use with compounding functions"""
|
"""Converts any time period to years for use with compounding functions"""
|
||||||
|
|
||||||
year_conversion_factor = {"years": 1, "months": 12, "days": 365}
|
year_conversion_factor = {"years": 1, "months": 12, "days": 365}
|
||||||
|
@ -112,6 +112,10 @@ class TestTimeSeriesCore:
|
|||||||
assert isinstance(ts, TimeSeriesCore)
|
assert isinstance(ts, TimeSeriesCore)
|
||||||
assert isinstance(ts, Mapping)
|
assert isinstance(ts, Mapping)
|
||||||
|
|
||||||
|
|
||||||
|
class TestSlicing:
|
||||||
|
data = [("2021-01-01", 220), ("2021-02-01", 230), ("2021-03-01", 240)]
|
||||||
|
|
||||||
def test_getitem(self):
|
def test_getitem(self):
|
||||||
ts = TimeSeriesCore(self.data, frequency="M")
|
ts = TimeSeriesCore(self.data, frequency="M")
|
||||||
assert ts.dates[0] == datetime.datetime(2021, 1, 1, 0, 0)
|
assert ts.dates[0] == datetime.datetime(2021, 1, 1, 0, 0)
|
||||||
@ -165,6 +169,15 @@ class TestTimeSeriesCore:
|
|||||||
assert dates[0] == datetime.datetime(2021, 1, 1, 0, 0)
|
assert dates[0] == datetime.datetime(2021, 1, 1, 0, 0)
|
||||||
assert values[0] == 220
|
assert values[0] == 220
|
||||||
|
|
||||||
|
def test_iloc_slicing(self):
|
||||||
|
ts = TimeSeriesCore(self.data, frequency="M")
|
||||||
|
assert ts.iloc[0] == (datetime.datetime(2021, 1, 1), 220)
|
||||||
|
assert ts.iloc[-1] == (datetime.datetime(2021, 3, 1), 240)
|
||||||
|
|
||||||
|
ts_slice = ts.iloc[0:2]
|
||||||
|
assert isinstance(ts_slice, TimeSeriesCore)
|
||||||
|
assert len(ts_slice) == 2
|
||||||
|
|
||||||
|
|
||||||
class TestTimeSeriesCoreHeadTail:
|
class TestTimeSeriesCoreHeadTail:
|
||||||
data = [
|
data = [
|
||||||
|
@ -1,82 +1,97 @@
|
|||||||
import datetime
|
import datetime
|
||||||
import os
|
import math
|
||||||
import random
|
import random
|
||||||
from typing import Literal, Sequence
|
from typing import List
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from fincal.core import Frequency, Series
|
from dateutil.relativedelta import relativedelta
|
||||||
|
from fincal.core import AllFrequencies, Frequency
|
||||||
from fincal.exceptions import DateNotFoundError
|
from fincal.exceptions import DateNotFoundError
|
||||||
from fincal.fincal import TimeSeries, create_date_series
|
from fincal.fincal import TimeSeries, create_date_series
|
||||||
from fincal.utils import FincalOptions
|
from fincal.utils import FincalOptions
|
||||||
|
|
||||||
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
||||||
sample_data_path = os.path.join(THIS_DIR, "data")
|
def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list:
|
||||||
|
"""Generates a price following a geometric brownian motion process based on the input of the arguments.
|
||||||
|
|
||||||
|
Since this function is used only to generate data for tests, the seed is fixed as 1234.
|
||||||
|
Many of the tests rely on exact values generated using this seed.
|
||||||
|
If the seed is changed, those tests will fail.
|
||||||
|
|
||||||
|
Parameters:
|
||||||
|
------------
|
||||||
|
s0: float
|
||||||
|
Asset inital price.
|
||||||
|
|
||||||
|
mu: float
|
||||||
|
Interest rate expressed annual terms.
|
||||||
|
|
||||||
|
sigma: float
|
||||||
|
Volatility expressed annual terms.
|
||||||
|
|
||||||
|
num_prices: int
|
||||||
|
number of prices to generate
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
--------
|
||||||
|
Returns a list of values generated using GBM algorithm
|
||||||
|
"""
|
||||||
|
|
||||||
|
random.seed(1234) # WARNING! Changing the seed will cause most tests to fail
|
||||||
|
all_values = []
|
||||||
|
for _ in range(num_prices):
|
||||||
|
s0 *= math.exp(
|
||||||
|
(mu - 0.5 * sigma**2) * (1.0 / 365.0) + sigma * math.sqrt(1.0 / 365.0) * random.gauss(mu=0, sigma=1)
|
||||||
|
)
|
||||||
|
all_values.append(round(s0, 2))
|
||||||
|
|
||||||
|
return all_values
|
||||||
|
|
||||||
|
|
||||||
def create_random_test_data(
|
def create_test_data(
|
||||||
frequency: str,
|
frequency: Frequency,
|
||||||
eomonth: bool,
|
num: int = 1000,
|
||||||
n: int,
|
skip_weekends: bool = False,
|
||||||
gaps: float,
|
mu: float = 0.1,
|
||||||
month_position: Literal["start", "middle", "end"],
|
sigma: float = 0.05,
|
||||||
date_as_str: bool,
|
eomonth: bool = False,
|
||||||
as_outer_type: Literal["dict", "list"] = "list",
|
) -> List[tuple]:
|
||||||
as_inner_type: Literal["dict", "list", "tuple"] = "tuple",
|
"""Creates TimeSeries data
|
||||||
) -> Sequence[tuple]:
|
|
||||||
start_dates = {
|
Parameters:
|
||||||
"start": datetime.datetime(2016, 1, 1),
|
-----------
|
||||||
"middle": datetime.datetime(2016, 1, 15),
|
frequency: Frequency
|
||||||
"end": datetime.datetime(2016, 1, 31),
|
The frequency of the time series data to be generated.
|
||||||
|
|
||||||
|
num: int
|
||||||
|
Number of date: value pairs to be generated.
|
||||||
|
|
||||||
|
skip_weekends: bool
|
||||||
|
Whether weekends (saturday, sunday) should be skipped.
|
||||||
|
Gets used only if the frequency is daily.
|
||||||
|
|
||||||
|
mu: float
|
||||||
|
Mean return for the values.
|
||||||
|
|
||||||
|
sigma: float
|
||||||
|
standard deviation of the values.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
--------
|
||||||
|
Returns a TimeSeries object
|
||||||
|
"""
|
||||||
|
|
||||||
|
start_date = datetime.datetime(2017, 1, 1)
|
||||||
|
timedelta_dict = {
|
||||||
|
frequency.freq_type: int(
|
||||||
|
frequency.value * num * (7 / 5 if frequency == AllFrequencies.D and skip_weekends else 1)
|
||||||
|
)
|
||||||
}
|
}
|
||||||
end_date = datetime.datetime(2021, 12, 31)
|
end_date = start_date + relativedelta(**timedelta_dict)
|
||||||
dates = create_date_series(start_dates[month_position], end_date, frequency=frequency, eomonth=eomonth)
|
dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends, eomonth=eomonth)
|
||||||
dates = dates[:n]
|
values = create_prices(1000, mu, sigma, num)
|
||||||
if gaps:
|
ts = list(zip(dates, values))
|
||||||
num_gaps = int(len(dates) * gaps)
|
return ts
|
||||||
to_remove = random.sample(dates, num_gaps)
|
|
||||||
for i in to_remove:
|
|
||||||
dates.remove(i)
|
|
||||||
if date_as_str:
|
|
||||||
dates = [i.strftime("%Y-%m-%d") for i in dates]
|
|
||||||
|
|
||||||
values = [random.randint(8000, 90000) / 100 for _ in dates]
|
|
||||||
|
|
||||||
data = list(zip(dates, values))
|
|
||||||
if as_outer_type == "list":
|
|
||||||
if as_inner_type == "list":
|
|
||||||
data = [list(i) for i in data]
|
|
||||||
elif as_inner_type == "dict[1]":
|
|
||||||
data = [dict((i,)) for i in data]
|
|
||||||
elif as_inner_type == "dict[2]":
|
|
||||||
data = [dict(date=i, value=j) for i, j in data]
|
|
||||||
elif as_outer_type == "dict":
|
|
||||||
data = dict(data)
|
|
||||||
|
|
||||||
return data
|
|
||||||
|
|
||||||
|
|
||||||
def create_organised_test_data() -> dict:
|
|
||||||
"""Creates organised test data so that output is exactly same in each run"""
|
|
||||||
|
|
||||||
all_dates, all_values = [], []
|
|
||||||
prev_date, prev_number = datetime.datetime(2018, 1, 1), 1000
|
|
||||||
|
|
||||||
for i in range(1, 1000):
|
|
||||||
if i % 5 == 0:
|
|
||||||
prev_date += datetime.timedelta(days=3)
|
|
||||||
else:
|
|
||||||
prev_date += datetime.timedelta(days=1)
|
|
||||||
all_dates.append(prev_date)
|
|
||||||
|
|
||||||
for i in range(1, 1000):
|
|
||||||
rem = i % 7
|
|
||||||
if rem % 2:
|
|
||||||
prev_number -= rem
|
|
||||||
else:
|
|
||||||
prev_number += rem
|
|
||||||
all_values.append(prev_number)
|
|
||||||
|
|
||||||
return dict(zip(all_dates, all_values))
|
|
||||||
|
|
||||||
|
|
||||||
class TestDateSeries:
|
class TestDateSeries:
|
||||||
@ -141,47 +156,70 @@ class TestDateSeries:
|
|||||||
assert datetime.datetime(2020, 11, 30) in d
|
assert datetime.datetime(2020, 11, 30) in d
|
||||||
|
|
||||||
|
|
||||||
class TestFincalBasic:
|
class TestTimeSeriesCreation:
|
||||||
def test_creation(self):
|
def test_creation_with_list_of_tuples(self):
|
||||||
data = create_random_test_data(
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50)
|
||||||
frequency="D", eomonth=False, n=50, gaps=0, month_position="start", date_as_str=True
|
ts = TimeSeries(ts_data, frequency="D")
|
||||||
)
|
assert len(ts) == 50
|
||||||
time_series = TimeSeries(data, frequency="D")
|
assert isinstance(ts.frequency, Frequency)
|
||||||
assert len(time_series) == 50
|
assert ts.frequency.days == 1
|
||||||
assert isinstance(time_series.frequency, Frequency)
|
|
||||||
assert time_series.frequency.days == 1
|
|
||||||
|
|
||||||
ffill_data = time_series.ffill()
|
def test_creation_with_string_dates(self):
|
||||||
assert len(ffill_data) == 50
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50)
|
||||||
|
ts_data1 = [(dt.strftime("%Y-%m-%d"), val) for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D")
|
||||||
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
|
|
||||||
data = create_random_test_data(
|
ts_data1 = [(dt.strftime("%d-%m-%Y"), val) for dt, val in ts_data]
|
||||||
frequency="D", eomonth=False, n=500, gaps=0.1, month_position="start", date_as_str=True
|
ts = TimeSeries(ts_data1, frequency="D", date_format="%d-%m-%Y")
|
||||||
)
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
time_series = TimeSeries(data, frequency="D")
|
|
||||||
assert len(time_series) == 450
|
|
||||||
|
|
||||||
|
ts_data1 = [(dt.strftime("%m-%d-%Y"), val) for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D", date_format="%m-%d-%Y")
|
||||||
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
|
|
||||||
|
ts_data1 = [(dt.strftime("%m-%d-%Y %H:%M"), val) for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D", date_format="%m-%d-%Y %H:%M")
|
||||||
|
datetime.datetime(2017, 1, 1, 0, 0) in ts
|
||||||
|
|
||||||
|
def test_creation_with_list_of_dicts(self):
|
||||||
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50)
|
||||||
|
ts_data1 = [{"date": dt.strftime("%Y-%m-%d"), "value": val} for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D")
|
||||||
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
|
|
||||||
|
def test_creation_with_list_of_lists(self):
|
||||||
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50)
|
||||||
|
ts_data1 = [[dt.strftime("%Y-%m-%d"), val] for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D")
|
||||||
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
|
|
||||||
|
def test_creation_with_dict(self):
|
||||||
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50)
|
||||||
|
ts_data1 = [{dt.strftime("%Y-%m-%d"): val} for dt, val in ts_data]
|
||||||
|
ts = TimeSeries(ts_data1, frequency="D")
|
||||||
|
datetime.datetime(2017, 1, 1) in ts
|
||||||
|
|
||||||
|
|
||||||
|
class TestTimeSeriesBasics:
|
||||||
def test_fill(self):
|
def test_fill(self):
|
||||||
data = create_random_test_data(
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50, skip_weekends=True)
|
||||||
frequency="D", eomonth=False, n=500, gaps=0.1, month_position="start", date_as_str=True
|
ts = TimeSeries(ts_data, frequency="D")
|
||||||
)
|
ffill_data = ts.ffill()
|
||||||
time_series = TimeSeries(data, frequency="D")
|
assert len(ffill_data) == 68
|
||||||
ffill_data = time_series.ffill()
|
|
||||||
assert len(ffill_data) >= 498
|
|
||||||
|
|
||||||
ffill_data = time_series.ffill(inplace=True)
|
ffill_data = ts.ffill(inplace=True)
|
||||||
assert ffill_data is None
|
assert ffill_data is None
|
||||||
assert len(time_series) >= 498
|
assert len(ts) == 68
|
||||||
|
|
||||||
data = create_random_test_data(
|
ts_data = create_test_data(frequency=AllFrequencies.D, num=50, skip_weekends=True)
|
||||||
frequency="D", eomonth=False, n=500, gaps=0.1, month_position="start", date_as_str=True
|
ts = TimeSeries(ts_data, frequency="D")
|
||||||
)
|
bfill_data = ts.bfill()
|
||||||
time_series = TimeSeries(data, frequency="D")
|
assert len(bfill_data) == 68
|
||||||
bfill_data = time_series.bfill()
|
|
||||||
assert len(bfill_data) >= 498
|
|
||||||
|
|
||||||
bfill_data = time_series.bfill(inplace=True)
|
bfill_data = ts.bfill(inplace=True)
|
||||||
assert bfill_data is None
|
assert bfill_data is None
|
||||||
assert len(time_series) >= 498
|
assert len(ts) == 68
|
||||||
|
|
||||||
data = [("2021-01-01", 220), ("2021-01-02", 230), ("2021-03-04", 240)]
|
data = [("2021-01-01", 220), ("2021-01-02", 230), ("2021-03-04", 240)]
|
||||||
ts = TimeSeries(data, frequency="D")
|
ts = TimeSeries(data, frequency="D")
|
||||||
@ -191,32 +229,77 @@ class TestFincalBasic:
|
|||||||
bf = ts.bfill()
|
bf = ts.bfill()
|
||||||
assert bf["2021-01-03"][1] == 240
|
assert bf["2021-01-03"][1] == 240
|
||||||
|
|
||||||
def test_iloc_slicing(self):
|
|
||||||
data = create_random_test_data(
|
|
||||||
frequency="D", eomonth=False, n=50, gaps=0, month_position="start", date_as_str=True
|
|
||||||
)
|
|
||||||
time_series = TimeSeries(data, frequency="D")
|
|
||||||
assert time_series.iloc[0] is not None
|
|
||||||
assert time_series.iloc[:3] is not None
|
|
||||||
assert time_series.iloc[5:7] is not None
|
|
||||||
assert isinstance(time_series.iloc[0], tuple)
|
|
||||||
assert isinstance(time_series.iloc[10:20], TimeSeries)
|
|
||||||
assert len(time_series.iloc[10:20]) == 10
|
|
||||||
|
|
||||||
def test_key_slicing(self):
|
|
||||||
data = create_random_test_data(
|
|
||||||
frequency="D", eomonth=False, n=50, gaps=0, month_position="start", date_as_str=True
|
|
||||||
)
|
|
||||||
time_series = TimeSeries(data, frequency="D")
|
|
||||||
available_date = time_series.iloc[5][0]
|
|
||||||
assert time_series[available_date] is not None
|
|
||||||
assert isinstance(time_series["dates"], Series)
|
|
||||||
assert isinstance(time_series["values"], Series)
|
|
||||||
assert len(time_series.dates) == 50
|
|
||||||
assert len(time_series.values) == 50
|
|
||||||
|
|
||||||
|
|
||||||
class TestReturns:
|
class TestReturns:
|
||||||
|
def test_returns_calc(self):
|
||||||
|
ts_data = create_test_data(AllFrequencies.D, skip_weekends=True)
|
||||||
|
ts = TimeSeries(ts_data, "D")
|
||||||
|
returns = ts.calculate_returns(
|
||||||
|
"2020-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_value=1
|
||||||
|
)
|
||||||
|
assert round(returns[1], 6) == 0.112913
|
||||||
|
|
||||||
|
returns = ts.calculate_returns(
|
||||||
|
"2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_value=3
|
||||||
|
)
|
||||||
|
assert round(returns[1], 6) == 0.015908
|
||||||
|
|
||||||
|
returns = ts.calculate_returns(
|
||||||
|
"2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_value=3
|
||||||
|
)
|
||||||
|
assert round(returns[1], 6) == 0.065167
|
||||||
|
|
||||||
|
returns = ts.calculate_returns(
|
||||||
|
"2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_value=90
|
||||||
|
)
|
||||||
|
assert round(returns[1], 6) == 0.017673
|
||||||
|
|
||||||
|
returns = ts.calculate_returns(
|
||||||
|
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
||||||
|
)
|
||||||
|
assert round(returns[1], 6) == 0.073632
|
||||||
|
|
||||||
|
with pytest.raises(DateNotFoundError):
|
||||||
|
ts.calculate_returns("2020-04-04", return_period_unit="days", return_period_value=90, as_on_match="exact")
|
||||||
|
with pytest.raises(DateNotFoundError):
|
||||||
|
ts.calculate_returns("2020-04-04", return_period_unit="months", return_period_value=3, prior_match="exact")
|
||||||
|
|
||||||
|
def test_date_formats(self):
|
||||||
|
ts_data = create_test_data(AllFrequencies.D, skip_weekends=True)
|
||||||
|
ts = TimeSeries(ts_data, "D")
|
||||||
|
FincalOptions.date_format = "%d-%m-%Y"
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
ts.calculate_returns(
|
||||||
|
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
||||||
|
)
|
||||||
|
|
||||||
|
returns1 = ts.calculate_returns(
|
||||||
|
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
|
||||||
|
)
|
||||||
|
returns2 = ts.calculate_returns("01-04-2020", return_period_unit="days", return_period_value=90)
|
||||||
|
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
|
||||||
|
|
||||||
|
FincalOptions.date_format = "%m-%d-%Y"
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
ts.calculate_returns(
|
||||||
|
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
||||||
|
)
|
||||||
|
|
||||||
|
returns1 = ts.calculate_returns(
|
||||||
|
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
|
||||||
|
)
|
||||||
|
returns2 = ts.calculate_returns("04-01-2020", return_period_unit="days", return_period_value=90)
|
||||||
|
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
|
||||||
|
|
||||||
|
def test_limits(self):
|
||||||
|
FincalOptions.date_format = "%Y-%m-%d"
|
||||||
|
ts_data = create_test_data(AllFrequencies.D)
|
||||||
|
ts = TimeSeries(ts_data, "D")
|
||||||
|
with pytest.raises(DateNotFoundError):
|
||||||
|
ts.calculate_returns("2020-11-25", return_period_unit="days", return_period_value=90, closest_max_days=10)
|
||||||
|
|
||||||
|
|
||||||
|
class TestReturnsAgain:
|
||||||
data = [
|
data = [
|
||||||
("2020-01-01", 10),
|
("2020-01-01", 10),
|
||||||
("2020-02-01", 12),
|
("2020-02-01", 12),
|
||||||
@ -298,13 +381,58 @@ class TestReturns:
|
|||||||
|
|
||||||
|
|
||||||
class TestVolatility:
|
class TestVolatility:
|
||||||
data = create_organised_test_data()
|
def test_daily_ts(self):
|
||||||
|
ts_data = create_test_data(AllFrequencies.D)
|
||||||
def test_volatility_basic(self):
|
ts = TimeSeries(ts_data, "D")
|
||||||
ts = TimeSeries(self.data, frequency="D")
|
assert len(ts) == 1000
|
||||||
sd = ts.volatility()
|
|
||||||
assert len(ts) == 999
|
|
||||||
assert round(sd, 6) == 0.057391
|
|
||||||
|
|
||||||
sd = ts.volatility(annualize_volatility=False)
|
sd = ts.volatility(annualize_volatility=False)
|
||||||
assert round(sd, 6) == 0.003004
|
assert round(sd, 6) == 0.002622
|
||||||
|
sd = ts.volatility()
|
||||||
|
assert round(sd, 6) == 0.050098
|
||||||
|
sd = ts.volatility(annual_compounded_returns=True)
|
||||||
|
assert round(sd, 4) == 37.9329
|
||||||
|
sd = ts.volatility(return_period_unit="months", annual_compounded_returns=True)
|
||||||
|
assert round(sd, 4) == 0.6778
|
||||||
|
sd = ts.volatility(return_period_unit="years")
|
||||||
|
assert round(sd, 6) == 0.023164
|
||||||
|
sd = ts.volatility(from_date="2017-10-01", to_date="2019-08-31", annualize_volatility=True)
|
||||||
|
assert round(sd, 6) == 0.050559
|
||||||
|
sd = ts.volatility(from_date="2017-02-01", frequency="M", return_period_unit="months")
|
||||||
|
assert round(sd, 6) == 0.050884
|
||||||
|
sd = ts.volatility(
|
||||||
|
frequency="M",
|
||||||
|
return_period_unit="months",
|
||||||
|
return_period_value=3,
|
||||||
|
annualize_volatility=False,
|
||||||
|
)
|
||||||
|
assert round(sd, 6) == 0.020547
|
||||||
|
|
||||||
|
|
||||||
|
class TestDrawdown:
|
||||||
|
def test_daily_ts(self):
|
||||||
|
ts_data = create_test_data(AllFrequencies.D, skip_weekends=True)
|
||||||
|
ts = TimeSeries(ts_data, "D")
|
||||||
|
mdd = ts.max_drawdown()
|
||||||
|
assert isinstance(mdd, dict)
|
||||||
|
assert len(mdd) == 3
|
||||||
|
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
|
||||||
|
expeced_response = {
|
||||||
|
"start_date": datetime.datetime(2017, 6, 6, 0, 0),
|
||||||
|
"end_date": datetime.datetime(2017, 7, 31, 0, 0),
|
||||||
|
"drawdown": -0.028293686030751997,
|
||||||
|
}
|
||||||
|
assert mdd == expeced_response
|
||||||
|
|
||||||
|
def test_weekly_ts(self):
|
||||||
|
ts_data = create_test_data(AllFrequencies.W, mu=1, sigma=0.5)
|
||||||
|
ts = TimeSeries(ts_data, "W")
|
||||||
|
mdd = ts.max_drawdown()
|
||||||
|
assert isinstance(mdd, dict)
|
||||||
|
assert len(mdd) == 3
|
||||||
|
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
|
||||||
|
expeced_response = {
|
||||||
|
"start_date": datetime.datetime(2019, 2, 17, 0, 0),
|
||||||
|
"end_date": datetime.datetime(2019, 11, 17, 0, 0),
|
||||||
|
"drawdown": -0.2584760499552089,
|
||||||
|
}
|
||||||
|
assert mdd == expeced_response
|
||||||
|
@ -1,210 +0,0 @@
|
|||||||
import datetime
|
|
||||||
import math
|
|
||||||
import random
|
|
||||||
from unittest import skip
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from dateutil.relativedelta import relativedelta
|
|
||||||
from fincal.core import AllFrequencies, Frequency
|
|
||||||
from fincal.exceptions import DateNotFoundError
|
|
||||||
from fincal.fincal import MaxDrawdown, TimeSeries, create_date_series
|
|
||||||
from fincal.utils import FincalOptions
|
|
||||||
|
|
||||||
|
|
||||||
def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list:
|
|
||||||
"""Generates a price following a geometric brownian motion process based on the input of the arguments.
|
|
||||||
|
|
||||||
Since this function is used only to generate data for tests, the seed is fixed as 1234.
|
|
||||||
Many of the tests rely on exact values generated using this seed.
|
|
||||||
If the seed is changed, those tests will fail.
|
|
||||||
|
|
||||||
Parameters:
|
|
||||||
------------
|
|
||||||
s0: float
|
|
||||||
Asset inital price.
|
|
||||||
|
|
||||||
mu: float
|
|
||||||
Interest rate expressed annual terms.
|
|
||||||
|
|
||||||
sigma: float
|
|
||||||
Volatility expressed annual terms.
|
|
||||||
|
|
||||||
num_prices: int
|
|
||||||
number of prices to generate
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
--------
|
|
||||||
Returns a list of values generated using GBM algorithm
|
|
||||||
"""
|
|
||||||
|
|
||||||
random.seed(1234) # WARNING! Changing the seed will cause most tests to fail
|
|
||||||
all_values = []
|
|
||||||
for _ in range(num_prices):
|
|
||||||
s0 *= math.exp(
|
|
||||||
(mu - 0.5 * sigma**2) * (1.0 / 365.0) + sigma * math.sqrt(1.0 / 365.0) * random.gauss(mu=0, sigma=1)
|
|
||||||
)
|
|
||||||
all_values.append(round(s0, 2))
|
|
||||||
|
|
||||||
return all_values
|
|
||||||
|
|
||||||
|
|
||||||
def create_test_timeseries(
|
|
||||||
frequency: Frequency, num: int = 1000, skip_weekends: bool = False, mu: float = 0.1, sigma: float = 0.05
|
|
||||||
) -> TimeSeries:
|
|
||||||
"""Creates TimeSeries data
|
|
||||||
|
|
||||||
Parameters:
|
|
||||||
-----------
|
|
||||||
frequency: Frequency
|
|
||||||
The frequency of the time series data to be generated.
|
|
||||||
|
|
||||||
num: int
|
|
||||||
Number of date: value pairs to be generated.
|
|
||||||
|
|
||||||
skip_weekends: bool
|
|
||||||
Whether weekends (saturday, sunday) should be skipped.
|
|
||||||
Gets used only if the frequency is daily.
|
|
||||||
|
|
||||||
mu: float
|
|
||||||
Mean return for the values.
|
|
||||||
|
|
||||||
sigma: float
|
|
||||||
standard deviation of the values.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
--------
|
|
||||||
Returns a TimeSeries object
|
|
||||||
"""
|
|
||||||
|
|
||||||
start_date = datetime.datetime(2017, 1, 1)
|
|
||||||
timedelta_dict = {
|
|
||||||
frequency.freq_type: int(
|
|
||||||
frequency.value * num * (7 / 5 if frequency == AllFrequencies.D and skip_weekends else 1)
|
|
||||||
)
|
|
||||||
}
|
|
||||||
end_date = start_date + relativedelta(**timedelta_dict)
|
|
||||||
dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends)
|
|
||||||
values = create_prices(1000, mu, sigma, num)
|
|
||||||
ts = TimeSeries(dict(zip(dates, values)), frequency=frequency.symbol)
|
|
||||||
return ts
|
|
||||||
|
|
||||||
|
|
||||||
class TestReturns:
|
|
||||||
def test_returns_calc(self):
|
|
||||||
ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
|
|
||||||
returns = ts.calculate_returns(
|
|
||||||
"2020-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_value=1
|
|
||||||
)
|
|
||||||
assert round(returns[1], 6) == 0.112913
|
|
||||||
|
|
||||||
returns = ts.calculate_returns(
|
|
||||||
"2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_value=3
|
|
||||||
)
|
|
||||||
assert round(returns[1], 6) == 0.015908
|
|
||||||
|
|
||||||
returns = ts.calculate_returns(
|
|
||||||
"2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_value=3
|
|
||||||
)
|
|
||||||
assert round(returns[1], 6) == 0.065167
|
|
||||||
|
|
||||||
returns = ts.calculate_returns(
|
|
||||||
"2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_value=90
|
|
||||||
)
|
|
||||||
assert round(returns[1], 6) == 0.017673
|
|
||||||
|
|
||||||
returns = ts.calculate_returns(
|
|
||||||
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
|
||||||
)
|
|
||||||
assert round(returns[1], 6) == 0.073632
|
|
||||||
|
|
||||||
with pytest.raises(DateNotFoundError):
|
|
||||||
ts.calculate_returns("2020-04-04", return_period_unit="days", return_period_value=90, as_on_match="exact")
|
|
||||||
with pytest.raises(DateNotFoundError):
|
|
||||||
ts.calculate_returns("2020-04-04", return_period_unit="months", return_period_value=3, prior_match="exact")
|
|
||||||
|
|
||||||
def test_date_formats(self):
|
|
||||||
ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
|
|
||||||
FincalOptions.date_format = "%d-%m-%Y"
|
|
||||||
with pytest.raises(ValueError):
|
|
||||||
ts.calculate_returns(
|
|
||||||
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
|
||||||
)
|
|
||||||
|
|
||||||
returns1 = ts.calculate_returns(
|
|
||||||
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
|
|
||||||
)
|
|
||||||
returns2 = ts.calculate_returns("01-04-2020", return_period_unit="days", return_period_value=90)
|
|
||||||
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
|
|
||||||
|
|
||||||
FincalOptions.date_format = "%m-%d-%Y"
|
|
||||||
with pytest.raises(ValueError):
|
|
||||||
ts.calculate_returns(
|
|
||||||
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
|
|
||||||
)
|
|
||||||
|
|
||||||
returns1 = ts.calculate_returns(
|
|
||||||
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
|
|
||||||
)
|
|
||||||
returns2 = ts.calculate_returns("04-01-2020", return_period_unit="days", return_period_value=90)
|
|
||||||
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
|
|
||||||
|
|
||||||
def test_limits(self):
|
|
||||||
FincalOptions.date_format = "%Y-%m-%d"
|
|
||||||
ts = create_test_timeseries(AllFrequencies.D)
|
|
||||||
with pytest.raises(DateNotFoundError):
|
|
||||||
ts.calculate_returns("2020-11-25", return_period_unit="days", return_period_value=90, closest_max_days=10)
|
|
||||||
|
|
||||||
|
|
||||||
class TestVolatility:
|
|
||||||
def test_daily_ts(self):
|
|
||||||
ts = create_test_timeseries(AllFrequencies.D)
|
|
||||||
assert len(ts) == 1000
|
|
||||||
sd = ts.volatility(annualize_volatility=False)
|
|
||||||
assert round(sd, 6) == 0.002622
|
|
||||||
sd = ts.volatility()
|
|
||||||
assert round(sd, 6) == 0.050098
|
|
||||||
sd = ts.volatility(annual_compounded_returns=True)
|
|
||||||
assert round(sd, 4) == 37.9329
|
|
||||||
sd = ts.volatility(return_period_unit="months", annual_compounded_returns=True)
|
|
||||||
assert round(sd, 4) == 0.6778
|
|
||||||
sd = ts.volatility(return_period_unit="years")
|
|
||||||
assert round(sd, 6) == 0.023164
|
|
||||||
sd = ts.volatility(from_date="2017-10-01", to_date="2019-08-31", annualize_volatility=True)
|
|
||||||
assert round(sd, 6) == 0.050559
|
|
||||||
sd = ts.volatility(from_date="2017-02-01", frequency="M", return_period_unit="months")
|
|
||||||
assert round(sd, 6) == 0.050884
|
|
||||||
sd = ts.volatility(
|
|
||||||
frequency="M",
|
|
||||||
return_period_unit="months",
|
|
||||||
return_period_value=3,
|
|
||||||
annualize_volatility=False,
|
|
||||||
)
|
|
||||||
assert round(sd, 6) == 0.020547
|
|
||||||
|
|
||||||
|
|
||||||
class TestDrawdown:
|
|
||||||
def test_daily_ts(self):
|
|
||||||
ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
|
|
||||||
mdd = ts.max_drawdown()
|
|
||||||
assert isinstance(mdd, dict)
|
|
||||||
assert len(mdd) == 3
|
|
||||||
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
|
|
||||||
expeced_response = {
|
|
||||||
"start_date": datetime.datetime(2017, 6, 6, 0, 0),
|
|
||||||
"end_date": datetime.datetime(2017, 7, 31, 0, 0),
|
|
||||||
"drawdown": -0.028293686030751997,
|
|
||||||
}
|
|
||||||
assert mdd == expeced_response
|
|
||||||
|
|
||||||
def test_weekly_ts(self):
|
|
||||||
ts = create_test_timeseries(AllFrequencies.W, mu=1, sigma=0.5)
|
|
||||||
mdd = ts.max_drawdown()
|
|
||||||
assert isinstance(mdd, dict)
|
|
||||||
assert len(mdd) == 3
|
|
||||||
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
|
|
||||||
expeced_response = {
|
|
||||||
"start_date": datetime.datetime(2019, 2, 17, 0, 0),
|
|
||||||
"end_date": datetime.datetime(2019, 11, 17, 0, 0),
|
|
||||||
"drawdown": -0.2584760499552089,
|
|
||||||
}
|
|
||||||
assert mdd == expeced_response
|
|
26
tests/test_utils.py
Normal file
26
tests/test_utils.py
Normal file
@ -0,0 +1,26 @@
|
|||||||
|
import datetime
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fincal.utils import _interval_to_years, _parse_date
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseDate:
|
||||||
|
def test_parsing(self):
|
||||||
|
dt = datetime.datetime(2020, 1, 1)
|
||||||
|
assert _parse_date(dt) == dt
|
||||||
|
assert _parse_date(dt.strftime("%Y-%m-%d")) == dt
|
||||||
|
assert _parse_date(datetime.date(2020, 1, 1)) == dt
|
||||||
|
assert _parse_date("01-01-2020", date_format="%d-%m-%Y") == dt
|
||||||
|
assert _parse_date("01-01-2020", date_format="%m-%d-%Y") == dt
|
||||||
|
|
||||||
|
def test_errors(self):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
_parse_date("01-01-2020")
|
||||||
|
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
_parse_date("abcdefg")
|
||||||
|
|
||||||
|
|
||||||
|
class TestIntervalToYears:
|
||||||
|
def test_months(self):
|
||||||
|
assert _interval_to_years("months", 6) == 0.5
|
Loading…
Reference in New Issue
Block a user