Run Black formatter on codebase (#68)

* Initial plan

* Add type hints, docstrings, and improve error handling

- Added comprehensive type hints to all functions in core.py, checks.py, datafeeds.py, strategies.py, and rules.py
- Added docstrings to all public functions explaining parameters and return values
- Improved error handling in csv_data() with specific exception types
- Removed IDE suppression comment in datafeeds.py
- Updated setup.py to reflect correct Python version requirement (3.8+)
- All tests pass successfully

Co-authored-by: michaelchu <540510+michaelchu@users.noreply.github.com>

* Fix code review comments: improve type hints consistency and error handling

- Fixed inconsistent use of List[tuple] to List[Tuple] throughout core.py
- Improved docstring accuracy for _remove_invalid_evaluated_options
- Enhanced error handling in csv_data to catch IndexError along with KeyError
- Updated docstrings to accurately reflect all exception types raised
- All tests continue to pass

Co-authored-by: michaelchu <540510+michaelchu@users.noreply.github.com>

* Run Black formatter on codebase

Applied Black formatter to ensure consistent code style across the repository.

Formatting changes:
- Split long function signatures across multiple lines for better readability
- Removed trailing whitespace from docstrings
- Applied consistent spacing throughout codebase

Files reformatted:
- optopsy/rules.py
- optopsy/checks.py
- optopsy/datafeeds.py
- optopsy/strategies.py
- optopsy/core.py
- tests/test_strategies.py

All 33 tests pass successfully after formatting.

Co-authored-by: michaelchu <540510+michaelchu@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: michaelchu <540510+michaelchu@users.noreply.github.com>

Author: Copilot

optopsy/checks.py

@@ -1,4 +1,7 @@
-expected_types = {
+from typing import Any, Callable, Dict, Tuple
+import pandas as pd
+
+expected_types: Dict[str, Tuple[str, ...]] = {
     "underlying_symbol": ("object", "str"),
     "underlying_price": ("int64", "float64"),
     "option_type": ("object", "str"),
@@ -10,44 +13,69 @@
 }
 
 
-def _run_checks(params, data):
+def _run_checks(params: Dict[str, Any], data: pd.DataFrame) -> None:
+    """
+    Run all validation checks on parameters and data.
+
+    Args:
+        params: Dictionary of strategy parameters
+        data: DataFrame containing option chain data
+
+    Raises:
+        ValueError: If any validation check fails
+    """
     for k, v in params.items():
         if k in param_checks:
             param_checks[k](k, v)
     _check_data_types(data)
 
 
-def _check_positive_integer(key, value):
+def _check_positive_integer(key: str, value: Any) -> None:
+    """Validate that value is a positive integer."""
     if value <= 0 or not isinstance(value, int):
         raise ValueError(f"Invalid setting for {key}, must be positive integer")
 
 
-def _check_positive_integer_inclusive(key, value):
+def _check_positive_integer_inclusive(key: str, value: Any) -> None:
+    """Validate that value is a non-negative integer (zero allowed)."""
     if value < 0 or not isinstance(value, int):
         raise ValueError(f"Invalid setting for {key}, must be positive integer, or 0")
 
 
-def _check_positive_float(key, value):
+def _check_positive_float(key: str, value: Any) -> None:
+    """Validate that value is a positive float."""
     if value <= 0 or not isinstance(value, float):
         raise ValueError(f"Invalid setting for {key}, must be positive float type")
 
 
-def _check_side(key, value):
+def _check_side(key: str, value: Any) -> None:
+    """Validate that value is either 'long' or 'short'."""
     if value != "long" and value != "short":
         raise ValueError(f"Invalid setting for '{key}', must be only 'long' or 'short'")
 
 
-def _check_bool_type(key, value):
+def _check_bool_type(key: str, value: Any) -> None:
+    """Validate that value is a boolean."""
     if not isinstance(value, bool):
         raise ValueError(f"Invalid setting for {key}, must be boolean type")
 
 
-def _check_list_type(key, value):
+def _check_list_type(key: str, value: Any) -> None:
+    """Validate that value is a list."""
     if not isinstance(value, list):
         raise ValueError(f"Invalid setting for {key}, must be a list type")
 
 
-def _check_data_types(data):
+def _check_data_types(data: pd.DataFrame) -> None:
+    """
+    Validate that DataFrame has required columns with correct data types.
+
+    Args:
+        data: DataFrame to validate
+
+    Raises:
+        ValueError: If required column is missing or has incorrect type
+    """
     df_type_dict = data.dtypes.astype(str).to_dict()
     for k, et in expected_types.items():
         if k not in df_type_dict:
@@ -58,7 +86,7 @@ def _check_data_types(data):
             )
 
 
-param_checks = {
+param_checks: Dict[str, Callable[[str, Any], None]] = {
     "dte_interval": _check_positive_integer,
     "max_entry_dte": _check_positive_integer,
     "exit_dte": _check_positive_integer_inclusive,

optopsy/core.py

@@ -1,3 +1,4 @@
+from typing import Any, Callable, Dict, List, Optional, Tuple
 import pandas as pd
 import numpy as np
 from functools import reduce
@@ -8,44 +9,57 @@
 pd.set_option("display.max_rows", None, "display.max_columns", None)
 
 
-def _assign_dte(data):
+def _assign_dte(data: pd.DataFrame) -> pd.DataFrame:
+    """Assign days to expiration (DTE) to the dataset."""
     return data.assign(dte=lambda r: (r["expiration"] - r["quote_date"]).dt.days)
 
 
-def _trim(data, col, lower, upper):
+def _trim(data: pd.DataFrame, col: str, lower: float, upper: float) -> pd.DataFrame:
+    """Filter dataframe rows where column value is between lower and upper bounds."""
     return data.loc[(data[col] >= lower) & (data[col] <= upper)]
 
 
-def _ltrim(data, col, lower):
+def _ltrim(data: pd.DataFrame, col: str, lower: float) -> pd.DataFrame:
+    """Filter dataframe rows where column value is greater than or equal to lower bound."""
     return data.loc[data[col] >= lower]
 
 
-def _rtrim(data, col, upper):
+def _rtrim(data: pd.DataFrame, col: str, upper: float) -> pd.DataFrame:
+    """Filter dataframe rows where column value is less than or equal to upper bound."""
     return data.loc[data[col] <= upper]
 
 
-def _get(data, col, val):
+def _get(data: pd.DataFrame, col: str, val: Any) -> pd.DataFrame:
+    """Filter dataframe rows where column equals specified value."""
     return data.loc[data[col] == val]
 
 
-def _remove_min_bid_ask(data, min_bid_ask):
+def _remove_min_bid_ask(data: pd.DataFrame, min_bid_ask: float) -> pd.DataFrame:
+    """Remove options with bid or ask prices below minimum threshold."""
     return data.loc[(data["bid"] > min_bid_ask) & (data["ask"] > min_bid_ask)]
 
 
-def _remove_invalid_evaluated_options(data):
+def _remove_invalid_evaluated_options(data: pd.DataFrame) -> pd.DataFrame:
+    """Keep evaluated options where entry DTE is greater than exit DTE."""
     return data.loc[
         (data["dte_exit"] <= data["dte_entry"])
         & (data["dte_entry"] != data["dte_exit"])
     ]
 
 
-def _cut_options_by_dte(data, dte_interval, max_entry_dte):
+def _cut_options_by_dte(
+    data: pd.DataFrame, dte_interval: int, max_entry_dte: int
+) -> pd.DataFrame:
+    """Categorize options into DTE intervals for grouping."""
     dte_intervals = list(range(0, max_entry_dte, dte_interval))
     data["dte_range"] = pd.cut(data["dte_entry"], dte_intervals)
     return data
 
 
-def _cut_options_by_otm(data, otm_pct_interval, max_otm_pct_interval):
+def _cut_options_by_otm(
+    data: pd.DataFrame, otm_pct_interval: float, max_otm_pct_interval: float
+) -> pd.DataFrame:
+    """Categorize options into out-of-the-money percentage intervals."""
     # consider using np.linspace in future
     otm_pct_intervals = [
         round(i, 2)
@@ -61,7 +75,10 @@ def _cut_options_by_otm(data, otm_pct_interval, max_otm_pct_interval):
     return data
 
 
-def _group_by_intervals(data, cols, drop_na):
+def _group_by_intervals(
+    data: pd.DataFrame, cols: List[str], drop_na: bool
+) -> pd.DataFrame:
+    """Group options by intervals and calculate descriptive statistics."""
     # this is a bottleneck, try to optimize
     grouped_dataset = data.groupby(cols)["pct_change"].describe()
 
@@ -73,8 +90,17 @@ def _group_by_intervals(data, cols, drop_na):
     return grouped_dataset
 
 
-def _evaluate_options(data, **kwargs):
+def _evaluate_options(data: pd.DataFrame, **kwargs: Any) -> pd.DataFrame:
+    """
+    Evaluate options by filtering, merging entry and exit data, and calculating costs.
 
+    Args:
+        data: DataFrame containing option chain data
+        **kwargs: Configuration parameters including max_otm_pct, min_bid_ask, exit_dte
+
+    Returns:
+        DataFrame with evaluated options including entry and exit prices
+    """
     # trim option chains with strikes too far out from current price
     data = data.pipe(_calculate_otm_pct).pipe(
         _trim,
@@ -103,7 +129,17 @@ def _evaluate_options(data, **kwargs):
     )[evaluated_cols]
 
 
-def _evaluate_all_options(data, **kwargs):
+def _evaluate_all_options(data: pd.DataFrame, **kwargs: Any) -> pd.DataFrame:
+    """
+    Complete pipeline to evaluate all options with DTE and OTM percentage categorization.
+
+    Args:
+        data: DataFrame containing option chain data
+        **kwargs: Configuration parameters for evaluation and categorization
+
+    Returns:
+        DataFrame with evaluated and categorized options
+    """
     return (
         data.pipe(_assign_dte)
         .pipe(_trim, "dte", kwargs["exit_dte"], kwargs["max_entry_dte"])
@@ -117,21 +153,25 @@ def _evaluate_all_options(data, **kwargs):
     )
 
 
-def _calls(data):
+def _calls(data: pd.DataFrame) -> pd.DataFrame:
+    """Filter dataframe for call options only."""
     return data[data.option_type.str.lower().str.startswith("c")]
 
 
-def _puts(data):
+def _puts(data: pd.DataFrame) -> pd.DataFrame:
+    """Filter dataframe for put options only."""
     return data[data.option_type.str.lower().str.startswith("p")]
 
 
-def _calculate_otm_pct(data):
+def _calculate_otm_pct(data: pd.DataFrame) -> pd.DataFrame:
+    """Calculate out-of-the-money percentage for each option."""
     return data.assign(
         otm_pct=lambda r: round((r["strike"] - r["underlying_price"]) / r["strike"], 2)
     )
 
 
-def _apply_ratios(data, leg_def):
+def _apply_ratios(data: pd.DataFrame, leg_def: List[Tuple]) -> pd.DataFrame:
+    """Apply position ratios (long/short multipliers) to entry and exit prices."""
     for idx in range(1, len(leg_def) + 1):
         entry_col = f"entry_leg{idx}"
         exit_col = f"exit_leg{idx}"
@@ -142,7 +182,10 @@ def _apply_ratios(data, leg_def):
     return data
 
 
-def _assign_profit(data, leg_def, suffixes):
+def _assign_profit(
+    data: pd.DataFrame, leg_def: List[Tuple], suffixes: List[str]
+) -> pd.DataFrame:
+    """Calculate total profit/loss and percentage change for multi-leg strategies."""
     data = _apply_ratios(data, leg_def)
 
     # determine all entry and exit columns
@@ -155,29 +198,48 @@ def _assign_profit(data, leg_def, suffixes):
 
     data["pct_change"] = np.where(
         data["total_entry_cost"].abs() > 0,
-        (data["total_exit_proceeds"] - data["total_entry_cost"]) / data["total_entry_cost"].abs(),
-        np.nan
+        (data["total_exit_proceeds"] - data["total_entry_cost"])
+        / data["total_entry_cost"].abs(),
+        np.nan,
     )
 
     return data
 
 
-def _strategy_engine(data, leg_def, join_on=None, rules=None):
+def _strategy_engine(
+    data: pd.DataFrame,
+    leg_def: List[Tuple],
+    join_on: Optional[List[str]] = None,
+    rules: Optional[Callable] = None,
+) -> pd.DataFrame:
+    """
+    Core strategy execution engine that constructs single or multi-leg option strategies.
+
+    Args:
+        data: DataFrame containing evaluated option data
+        leg_def: List of tuples defining strategy legs (side, filter_function)
+        join_on: Columns to join on for multi-leg strategies
+        rules: Optional filtering rules to apply after joining legs
+
+    Returns:
+        DataFrame with constructed strategy and calculated profit/loss
+    """
     if len(leg_def) == 1:
         data["pct_change"] = np.where(
             data["entry"].abs() > 0,
             (data["exit"] - data["entry"]) / data["entry"].abs(),
-            np.nan
+            np.nan,
         )
         return leg_def[0][1](data)
 
-    def _rule_func(d, r, ld):
+    def _rule_func(
+        d: pd.DataFrame, r: Optional[Callable], ld: List[Tuple]
+    ) -> pd.DataFrame:
         return d if r is None else r(d, ld)
 
     partials = [leg[1](data) for leg in leg_def]
     suffixes = [f"_leg{idx}" for idx in range(1, len(leg_def) + 1)]
 
-    # noinspection PyTypeChecker
     return (
         reduce(
             lambda left, right: pd.merge(
@@ -190,7 +252,17 @@ def _rule_func(d, r, ld):
     )
 
 
-def _process_strategy(data, **context):
+def _process_strategy(data: pd.DataFrame, **context: Any) -> pd.DataFrame:
+    """
+    Main entry point for processing option strategies.
+
+    Args:
+        data: DataFrame containing raw option chain data
+        **context: Dictionary containing strategy parameters, leg definitions, and formatting options
+
+    Returns:
+        DataFrame with processed strategy results
+    """
     _run_checks(context["params"], data)
     return (
         _evaluate_all_options(
@@ -217,7 +289,24 @@ def _process_strategy(data, **context):
     )
 
 
-def _format_output(data, params, internal_cols, external_cols):
+def _format_output(
+    data: pd.DataFrame,
+    params: Dict[str, Any],
+    internal_cols: List[str],
+    external_cols: List[str],
+) -> pd.DataFrame:
+    """
+    Format strategy output as either raw data or grouped statistics.
+
+    Args:
+        data: DataFrame with strategy results
+        params: Parameters including 'raw' and 'drop_nan' flags
+        internal_cols: Columns to include in raw output
+        external_cols: Columns to group by for statistics output
+
+    Returns:
+        Formatted DataFrame with either raw data or descriptive statistics
+    """
     if params["raw"]:
         return data[internal_cols].reset_index(drop=True)