Plugin Development

Build and ship your own FinFacts plugins using the lightweight finfacts-sdk and the headless CLI. This guide is self-contained: contract, sample plugin, sample data, and run/validate commands.

Install the SDK

• Clone the repo and install the SDK only: pip install .
• Exposed helpers: FieldSpec, CalcResult, CalcPlugin, build_nav_sample, ensure_pandas, default_plugins_dir.
• Optional dev extras: pip install .[dev]

Plugin contract (essentials)

• Class attributes: id, name
• Required methods:
• mapping_spec() -> List[FieldSpec] (declares expected columns)
• compute(dataframe, mapping, cancel_check=None) -> CalcResult (performs the calculation; poll cancel_check() in long loops)
• Optional methods:
• build_summary_spec(...) for custom summary rendering
• generate_sample(...) to provide realistic demo data
• CalcResult should populate kpis (dict) plus subperiod_headers and subperiod_rows for summary/exports.
• FieldSpec fields: id, label, required, kind (date|number|string), desc.

Directory layout

• Drop .py plugins into ~/.finfacts/plugins (or the path from finfacts_sdk.default_plugins_dir()).
• During development, point the CLI at the folder containing your plugin file.
• Keep plugins free of private apps.* imports so they work without the repo.

Sample plugin (copy/paste)

Save the code below as plugin_sample_net_return.py in an empty folder.

from datetime import datetime
from typing import Dict, List
from finfacts_sdk.types import CalcResult, FieldSpec
from finfacts_sdk.samples import build_nav_sample, ensure_pandas


class SampleNetReturnPlugin:
    """Minimal Dietz-style return plugin; copy and adapt for your use case."""

    id = "sample_net_return"
    name = "Sample Net Return (SDK Demo)"

    def mapping_spec(self) -> List[FieldSpec]:
        """Declare required/optional input fields."""
        return [
            FieldSpec("date", "Date", True, "date", "Date column (YYYY-MM-DD)."),
            FieldSpec("mv", "MarketValue", True, "number", "Portfolio market value or NAV."),
            FieldSpec("flow", "ExternalFlow", False, "number", "External cash flow signed to portfolio."),
        ]

    def compute(self, dataframe, mapping: Dict[str, str], cancel_check=None) -> CalcResult:
        """Compute period returns and an equity index from NAV + flows."""
        df = ensure_pandas(dataframe, mapping).copy()
        date_col = mapping.get("date", "Date")
        mv_col = mapping.get("mv", "MarketValue")
        flow_col = mapping.get("flow", "ExternalFlow")

        if date_col not in df.columns or mv_col not in df.columns:
            return CalcResult(kpis={}, subperiod_headers=[], subperiod_rows=[])

        df[date_col] = df[date_col].apply(lambda v: datetime.fromisoformat(str(v)).date())
        df = df.dropna(subset=[date_col, mv_col]).sort_values(date_col)
        if flow_col not in df.columns:
            df[flow_col] = 0

        headers = ["PeriodStart", "PeriodEnd", "Begin MV", "Net Flows", "End MV", "Return", "Cum Index"]
        rows: List[List[object]] = []
        eq_index = 1.0

        for i in range(1, len(df)):
            if cancel_check and cancel_check():
                return CalcResult(kpis={}, subperiod_headers=headers, subperiod_rows=rows)
            start_row = df.iloc[i - 1]
            end_row = df.iloc[i]
            bmv = float(start_row[mv_col])
            emv = float(end_row[mv_col])
            flow = float(end_row.get(flow_col, 0.0))
            denom = bmv + (flow / 2.0) if (bmv + (flow / 2.0)) != 0 else 1.0
            r = (emv - bmv - flow) / denom
            eq_index *= 1 + r
            rows.append(
                [
                    start_row[date_col],
                    end_row[date_col],
                    bmv,
                    flow,
                    emv,
                    r,
                    eq_index,
                ]
            )

        kpis = {
            "method": "sample_modified_dietz_style",
            "periods": str(len(rows)),
            "calc_version": "0.1.0",
        }
        return CalcResult(kpis=kpis, subperiod_headers=headers, subperiod_rows=rows)

    def generate_sample(self, *, years: int = 1, rows: int | None = None, freq_label: str = "monthly"):
        """Provide a small demo dataset when users ask for sample data."""
        periods = rows - 1 if rows else max(years * 12, 2)
        return build_nav_sample(periods=periods)

Sample data (paste into sample_nav.csv)

Date	MarketValue	ExternalFlow
2023-01-31	1000000	0
2023-02-28	1004000	10000
2023-03-31	1012000	10000
2023-04-30	1020000	-5000
2023-05-31	1031000	15000
2023-06-30	1038000	0
2023-07-31	1045000	0
2023-08-31	1056000	12000
2023-09-30	1059000	0
2023-10-31	1068000	-8000
2023-11-30	1075000	0
2023-12-31	1082000	10000
2024-01-31	1090000	0

Run the sample

finfacts-cli run \
  --plugins-dir . \
  --plugin-id sample_net_return \
  --input sample_nav.csv \
  --mapping '{"date":"Date","mv":"MarketValue","flow":"ExternalFlow"}' \
  --freq monthly \
  --output-dir /tmp/finfacts-sample-run

Validate and iterate

• Validate mappings without full runs: finfacts-cli validate --plugin-id sample_net_return --mapping '{"date":"Date","mv":"MarketValue"}' --input sample_nav.csv
• Add unit tests around compute; the SDK ships py.typed for mypy/pyright.
• Keep dependencies minimal to simplify distribution.

Distribution tips

• Ship a single .py file that users copy into their plugins directory, or publish a pip package.
• Document plugin id, version, minimum FinFacts version, and license in your README.
• Avoid bundling secrets or hard-coded credentials in plugin files.