Code refactoring

Author: andrewmurphy353

curo/daycount/convention.py

@@ -0,0 +1,45 @@
+"""_summary_
+
+Raises:
+    NotImplementedError: _description_
+
+Returns:
+    _type_: _description_
+"""
+
+import pandas as pd
+from daycount.day_count_factor import DayCountFactor
+from enums import DayCountOrigin
+
+class Convention:
+    """_summary_
+    """
+    def __init__(self, use_post_dates: bool = False, include_non_financing_flows: bool = False,
+                 use_xirr_method: bool = False):
+        self.use_post_dates = use_post_dates
+        self.include_non_financing_flows = include_non_financing_flows
+        self.use_xirr_method = use_xirr_method
+
+    def compute_factor(self, start: pd.Timestamp, end: pd.Timestamp) -> DayCountFactor:
+        """_summary_
+
+        Args:
+            start (pd.Timestamp): _description_
+            end (pd.Timestamp): _description_
+
+        Raises:
+            NotImplementedError: _description_
+
+        Returns:
+            DayCountFactor: _description_
+        """
+        raise NotImplementedError
+
+    @property
+    def day_count_origin(self) -> DayCountOrigin:
+        """_summary_
+
+        Returns:
+            DayCountOrigin: _description_
+        """
+        return DayCountOrigin.DRAWDOWN if self.use_xirr_method else DayCountOrigin.NEIGHBOUR

curo/daycount/day_count_factor.py

@@ -0,0 +1,58 @@
+"""_summary_
+"""
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+@dataclass
+class DayCountFactor:
+    """_summary_
+
+    Returns:
+        _type_: _description_
+    """
+    principal_factor: float
+    fractional_adjustment: Optional[float]
+    principal_operand_log: List[str]
+    fractional_operand_log: Optional[List[str]]
+
+    @classmethod
+    def us_appendix_j(cls, principal_factor: float,
+                      fractional_adjustment: float,
+                      principal_operand_log: List[str],
+                      fractional_operand_log: List[str]) -> 'DayCountFactor':
+        """_summary_
+
+        Args:
+            principal_factor (float): _description_
+            fractional_adjustment (float): _description_
+            principal_operand_log (List[str]): _description_
+            fractional_operand_log (List[str]): _description_
+
+        Returns:
+            DayCountFactor: _description_
+        """
+        return cls(principal_factor, fractional_adjustment, principal_operand_log,
+                   fractional_operand_log)
+
+    @property
+    def total_factor(self) -> float:
+        """_summary_
+
+        Returns:
+            float: _description_
+        """
+        return self.principal_factor + (self.fractional_adjustment or 0.0)
+
+    @staticmethod
+    def operands_to_string(numerator: int, denominator: int) -> str:
+        """_summary_
+
+        Args:
+            numerator (int): _description_
+            denominator (int): _description_
+
+        Returns:
+            str: _description_
+        """
+        return f"{numerator}/{denominator}"

curo/daycount/us_30_360.py

@@ -0,0 +1,48 @@
+"""_summary_
+
+Returns:
+    _type_: _description_
+"""
+
+import pandas as pd
+from convention import Convention
+from day_count_factor import DayCountFactor
+
+
+class US30360(Convention):
+    """_summary_
+
+    Args:
+        Convention (_type_): _description_
+    """
+    def __init__(self, use_post_dates: bool = True,
+                 include_non_financing_flows: bool = False,
+                 use_xirr_method: bool = False):
+        super().__init__(use_post_dates=use_post_dates,
+                         include_non_financing_flows=include_non_financing_flows,
+                         use_xirr_method=use_xirr_method)
+
+    def compute_factor(self, start: pd.Timestamp, end: pd.Timestamp) -> DayCountFactor:
+        dd1, mm1, yyyy1 = start.day, start.month, start.year
+        dd2, mm2, yyyy2 = end.day, end.month, end.year
+
+        z = 30 if dd1 == 31 else dd1
+        dt1 = 360 * yyyy1 + 30 * mm1 + z
+
+        if dd2 == 31 and (dd1 == 30 or dd1 == 31):
+            z = 30
+        elif dd2 == 31 and dd1 < 30:
+            z = dd2
+        else:
+            z = dd2
+        dt2 = 360 * yyyy2 + 30 * mm2 + z
+
+        numerator = abs(dt2 - dt1)
+        factor = numerator / 360
+
+        return DayCountFactor(
+            principal_factor=factor,
+            fractional_adjustment=None,
+            principal_operand_log=[DayCountFactor.operands_to_string(numerator, 360)],
+            fractional_operand_log=None
+        )

curo/daycount/us_30u_360.py

@@ -0,0 +1,60 @@
+"""
+Docstring for daycount.us_30u_360
+"""
+
+import pandas as pd
+from convention import Convention
+from day_count_factor import DayCountFactor
+from utils import is_leap_year
+
+class US30U360(Convention):
+    """_summary_
+
+    Args:
+        Convention (_type_): _description_
+    """
+    def __init__(self, use_post_dates: bool = True,
+                 include_non_financing_flows: bool = False,
+                 use_xirr_method: bool = False):
+        super().__init__(use_post_dates=use_post_dates,
+                         include_non_financing_flows=include_non_financing_flows,
+                         use_xirr_method=use_xirr_method)
+
+    def compute_factor(self, start: pd.Timestamp, end: pd.Timestamp) -> DayCountFactor:
+        dt1 = start.replace(day=30 if start.day == 31 else start.day)
+        dt2 = end.replace(day=30 if end.day == 31 else end.day)
+        days_diff = self._day_difference(dt1, dt2)
+        days = ((dt2.year - dt1.year) * 360) + ((dt2.month - dt1.month) * 30) + days_diff
+
+        numerator = abs(days)
+        factor = numerator / 360
+
+        return DayCountFactor(
+            principal_factor=factor,
+            fractional_adjustment=None,
+            principal_operand_log=[DayCountFactor.operands_to_string(numerator, 360)],
+            fractional_operand_log=None
+        )
+
+    def _day_difference(self, d1: pd.Timestamp, d2: pd.Timestamp) -> int:
+        is_d1_last_day = self._is_last_day_of_month(d1)
+        is_d2_last_day = self._is_last_day_of_month(d2)
+
+        if is_d1_last_day and is_d2_last_day:
+            return d2.day + (d1.day - d2.day) - d1.day
+        if is_d1_last_day and d1.month == 2 and d1.day != d2.day:
+            if not is_leap_year(d1.year) and d2.day == 29:
+                return d2.day - (d1.day + (29 - d1.day))
+            return d2.day - (d1.day + (30 - d1.day))
+        if is_d2_last_day and d2.month == 2 and d2.day != d1.day:
+            if not is_leap_year(d2.year) and d1.day == 29:
+                return (d2.day + (29 - d2.day)) - d1.day
+            return (d2.day + (30 - d2.day)) - d1.day
+        return d2.day - d1.day
+
+    def _is_last_day_of_month(self, date: pd.Timestamp) -> bool:
+        next_month = date + pd.offsets.MonthBegin(1)
+        end_of_month = next_month - pd.Timedelta(days=1)
+        return (date.year == end_of_month.year and
+                date.month == end_of_month.month and
+                date.day == end_of_month.day)

curo/daycount/us_appendix_j.py

@@ -0,0 +1,126 @@
+"""_summary_
+
+Raises:
+    ValueError: _description_
+
+Returns:
+    _type_: _description_
+"""
+
+from typing import List
+import pandas as pd
+from daycount.convention import Convention
+from daycount.day_count_factor import DayCountFactor
+from enums import DayCountTimePeriod
+from utils import roll_day, roll_month, has_month_end_day
+
+
+class USAppendixJ(Convention):
+    """_summary_
+
+    Args:
+        Convention (_type_): _description_
+    """
+    def __init__(self, time_period: DayCountTimePeriod = DayCountTimePeriod.MONTH):
+        """_summary_
+
+        Args:
+            time_period (DayCountTimePeriod, optional): _description_. 
+            Defaults to DayCountTimePeriod.MONTH.
+        """
+        super().__init__(use_post_dates=True,
+                         include_non_financing_flows=True,
+                         use_xirr_method=True)
+        self.time_period = time_period
+
+    def compute_factor(self, start: pd.Timestamp, end: pd.Timestamp) -> DayCountFactor:
+        """_summary_
+
+        Args:
+            start (pd.Timestamp): _description_
+            end (pd.Timestamp): _description_
+
+        Raises:
+            ValueError: _description_
+
+        Returns:
+            DayCountFactor: _description_
+        """
+        whole_periods = 0
+        initial_drawdown = start
+        start_whole_period = end
+        principal_operand_log: List[str] = []
+        fractional_operand_log: List[str] = []
+
+        if self.time_period == DayCountTimePeriod.DAY:
+            days = (start_whole_period - initial_drawdown).days
+            fractional_adjustment = days / 365.0
+            fractional_operand_log.append(DayCountFactor.operands_to_string(days, 365))
+            return DayCountFactor.us_appendix_j(
+                principal_factor=0.0,
+                fractional_adjustment=fractional_adjustment,
+                principal_operand_log=principal_operand_log,
+                fractional_operand_log=fractional_operand_log
+            )
+
+        while True:
+            temp_date = start_whole_period
+            if self.time_period == DayCountTimePeriod.YEAR:
+                temp_date = roll_month(start_whole_period, -12, end.day)
+            elif self.time_period == DayCountTimePeriod.HALF_YEAR:
+                temp_date = roll_month(start_whole_period, -6, end.day)
+            elif self.time_period == DayCountTimePeriod.QUARTER:
+                temp_date = roll_month(start_whole_period, -3, end.day)
+            elif self.time_period == DayCountTimePeriod.MONTH:
+                temp_date = roll_month(start_whole_period, -1, end.day)
+            elif self.time_period == DayCountTimePeriod.FORTNIGHT:
+                temp_date = roll_day(start_whole_period, -14)
+            elif self.time_period == DayCountTimePeriod.WEEK:
+                temp_date = roll_day(start_whole_period, -7)
+            else:
+                raise ValueError("Unsupported time period")
+
+            if not initial_drawdown > temp_date:
+                start_whole_period = temp_date
+                whole_periods += 1
+            else:
+                if self.time_period in [DayCountTimePeriod.YEAR, DayCountTimePeriod.HALF_YEAR,
+                                       DayCountTimePeriod.QUARTER, DayCountTimePeriod.MONTH]:
+                    if initial_drawdown.day == temp_date.day:
+                        break
+                    if (initial_drawdown.month == temp_date.month and
+                        has_month_end_day(initial_drawdown) and
+                        has_month_end_day(end)):
+                        if initial_drawdown.day <= temp_date.day or has_month_end_day(end):
+                            start_whole_period = initial_drawdown
+                            whole_periods += 1
+                break
+
+        principal_factor = float(whole_periods) if whole_periods > 0 else 0.0
+        if whole_periods > 0:
+            principal_operand_log.append(str(whole_periods))
+
+        fractional_adjustment = 0.0
+        if not initial_drawdown > start_whole_period:
+            days = (start_whole_period - initial_drawdown).days
+            denominator = {
+                DayCountTimePeriod.YEAR: 365,
+                DayCountTimePeriod.HALF_YEAR: 180,
+                DayCountTimePeriod.QUARTER: 90,
+                DayCountTimePeriod.MONTH: 30,
+                DayCountTimePeriod.FORTNIGHT: 15,
+                DayCountTimePeriod.WEEK: 7
+            }.get(self.time_period, 30)
+
+            if days > 0:
+                fractional_adjustment = days / denominator
+                fractional_operand_log.append(
+                    DayCountFactor.operands_to_string(days, denominator)
+                )
+
+        return DayCountFactor.us_appendix_j(
+            principal_factor=principal_factor,
+            fractional_adjustment=fractional_adjustment,
+            principal_operand_log=principal_operand_log,
+            fractional_operand_log=fractional_operand_log
+        )

curo/example.py

@@ -0,0 +1,42 @@
+"""
+Docstring for examples.example
+"""
+
+import pandas as pd
+#from curo import Calculator, SeriesAdvance
+#, SeriesPayment
+from calculator import Calculator
+from enums import Frequency, Mode
+from series import SeriesAdvance, SeriesPayment
+
+calculator = Calculator()
+
+# Step 2: Define the advance, payment, and/or charge series
+#calculator.add(
+#    series=SeriesAdvance(number_of=2, value=10000)
+#)
+
+series_payment = SeriesPayment(
+    number_of=3,
+   # post_date_from=pd.Timestamp('2025-01-01', tz='UTC'),
+)
+
+pd_dataframe = series_payment.to_cash_flows(pd.Timestamp('2025-01-27', tz='UTC'))
+
+print(pd_dataframe.to_records)
+
+
+# calculator.add(
+#     SeriesPayment(
+#         numberOf: 6,
+#         label: 'Instalment',
+#         value: null,
+#         mode: Mode.arrear,
+#     ),
+# )
+
+#   // 3. Calculate the unknown cash flow value
+#   final valueResult = await calculator.solveValue(
+#     dayCount: const US30360(),
+#     interestRate: 0.0825,
+#   );