Skip to content

Enterprise API Reference

Multi-Asset

financelib.enterprise.multi_asset.Asset(symbol: str, asset_class: AssetClass, name: str = '', currency: str = 'USD', exchange: str = '', metadata: Dict[str, Any] = dict()) dataclass

Unified asset representation across all asset classes.

Parameters:

Name Type Description Default
symbol str

Unique identifier (e.g., 'THYAO.IS', 'EURUSD', 'BTC/USDT').

 required
asset_class AssetClass

Type of asset.

 required
name str

Human-readable name.

 ''
currency str

Denomination currency.

 'USD'
exchange str

Primary exchange.

 ''
metadata Dict[str, Any]

Additional asset-specific data.

 dict()
Example

stock = Asset("THYAO.IS", AssetClass.EQUITY, "Turkish Airlines", "TRY", "BIST") bond = Asset("TR10Y", AssetClass.BOND, "Turkey 10Y Gov Bond", "TRY") fx = Asset("USDTRY", AssetClass.FX, "USD/TRY", "TRY")

financelib.enterprise.multi_asset.AssetClass

Bases: Enum

Supported asset classes.

financelib.enterprise.multi_asset.AssetUniverse()

Manages a collection of assets across all classes.

Provides filtering, grouping, and lookup by symbol or asset class.

Example

universe = AssetUniverse() universe.add(Asset("THYAO.IS", AssetClass.EQUITY, "THY", "TRY", "BIST")) universe.add(Asset("USDTRY", AssetClass.FX, "USD/TRY", "TRY")) equities = universe.filter_by_class(AssetClass.EQUITY)

Source code in financelib/enterprise/multi_asset.py 
89
90
def __init__(self) -> None:
    self._assets: Dict[str, Asset] = {}

Functions

add(asset: Asset) -> None

Add an asset to the universe.

Source code in financelib/enterprise/multi_asset.py 
92
93
94
def add(self, asset: Asset) -> None:
    """Add an asset to the universe."""
    self._assets[asset.symbol] = asset

add_many(assets: List[Asset]) -> None

Add multiple assets.

Source code in financelib/enterprise/multi_asset.py 
96
97
98
99
def add_many(self, assets: List[Asset]) -> None:
    """Add multiple assets."""
    for a in assets:
        self._assets[a.symbol] = a

filter_by_class(asset_class: AssetClass) -> List[Asset]

Get all assets of a specific class.

Source code in financelib/enterprise/multi_asset.py 
112
113
114
def filter_by_class(self, asset_class: AssetClass) -> List[Asset]:
    """Get all assets of a specific class."""
    return [a for a in self._assets.values() if a.asset_class == asset_class]

filter_by_currency(currency: str) -> List[Asset]

Get all assets denominated in a specific currency.

Source code in financelib/enterprise/multi_asset.py 
116
117
118
def filter_by_currency(self, currency: str) -> List[Asset]:
    """Get all assets denominated in a specific currency."""
    return [a for a in self._assets.values() if a.currency == currency.upper()]

filter_by_exchange(exchange: str) -> List[Asset]

Get all assets on a specific exchange.

Source code in financelib/enterprise/multi_asset.py 
120
121
122
def filter_by_exchange(self, exchange: str) -> List[Asset]:
    """Get all assets on a specific exchange."""
    return [a for a in self._assets.values() if a.exchange == exchange]

get(symbol: str) -> Optional[Asset]

Get asset by symbol.

Source code in financelib/enterprise/multi_asset.py 
101
102
103
def get(self, symbol: str) -> Optional[Asset]:
    """Get asset by symbol."""
    return self._assets.get(symbol)

group_by_class() -> Dict[str, List[Asset]]

Group all assets by asset class.

Source code in financelib/enterprise/multi_asset.py 
124
125
126
127
128
129
130
def group_by_class(self) -> Dict[str, List[Asset]]:
    """Group all assets by asset class."""
    groups: Dict[str, List[Asset]] = {}
    for a in self._assets.values():
        key = a.asset_class.value
        groups.setdefault(key, []).append(a)
    return groups

remove(symbol: str) -> bool

Remove an asset. Returns True if existed.

Source code in financelib/enterprise/multi_asset.py 
105
106
107
108
109
110
def remove(self, symbol: str) -> bool:
    """Remove an asset. Returns True if existed."""
    if symbol in self._assets:
        del self._assets[symbol]
        return True
    return False

summary() -> Dict[str, int]

Count of assets per class.

Source code in financelib/enterprise/multi_asset.py 
140
141
142
143
144
145
146
def summary(self) -> Dict[str, int]:
    """Count of assets per class."""
    counts: Dict[str, int] = {}
    for a in self._assets.values():
        key = a.asset_class.value
        counts[key] = counts.get(key, 0) + 1
    return counts

Order Routing

financelib.enterprise.order_router.OrderRouter()

Smart order router for multi-exchange trading.

Routes orders to the best available exchange based on asset class, exchange priority, and availability.

Example

router = OrderRouter() router.add_exchange(Exchange("BIST", ["equity"], priority=1)) router.add_exchange(Exchange("BINANCE", ["crypto"], priority=1)) order = router.route("THYAO.IS", "buy", 100, asset_class="equity")

Source code in financelib/enterprise/order_router.py 
 98
 99
100
101
def __init__(self) -> None:
    self._exchanges: Dict[str, Exchange] = {}
    self._order_counter = 0
    self._order_history: List[RoutedOrder] = []

Functions

add_exchange(exchange: Exchange) -> None

Register an exchange.

Source code in financelib/enterprise/order_router.py 
103
104
105
106
def add_exchange(self, exchange: Exchange) -> None:
    """Register an exchange."""
    self._exchanges[exchange.name] = exchange
    logger.info(f"Exchange registered: {exchange.name} ({exchange.asset_classes})")

find_exchanges(asset_class: str) -> List[Exchange]

Find active exchanges supporting an asset class, sorted by priority.

Source code in financelib/enterprise/order_router.py 
117
118
119
120
121
122
123
def find_exchanges(self, asset_class: str) -> List[Exchange]:
    """Find active exchanges supporting an asset class, sorted by priority."""
    matching = [
        e for e in self._exchanges.values()
        if asset_class in e.asset_classes and e.is_active
    ]
    return sorted(matching, key=lambda e: e.priority)

route(symbol: str, side: str, quantity: float, price: float = 0.0, order_type: str = 'market', asset_class: str = 'equity', preferred_exchange: Optional[str] = None) -> RoutedOrder

Route an order to the best available exchange.

Parameters:

Name Type Description Default
symbol str

Asset symbol.

 required
side str

'buy' or 'sell'.

 required
quantity float

Order quantity.

 required
price float

Limit price (0 for market).

 0.0
order_type str

'market', 'limit', or 'stop'.

 'market'
asset_class str

Asset class for routing.

 'equity'
preferred_exchange Optional[str]

Override automatic routing.

 None

Returns:

Type Description
RoutedOrder

RoutedOrder with exchange assignment and status.
Source code in financelib/enterprise/order_router.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def route(
    self,
    symbol: str,
    side: str,
    quantity: float,
    price: float = 0.0,
    order_type: str = "market",
    asset_class: str = "equity",
    preferred_exchange: Optional[str] = None,
) -> RoutedOrder:
    """Route an order to the best available exchange.

    Args:
        symbol: Asset symbol.
        side: 'buy' or 'sell'.
        quantity: Order quantity.
        price: Limit price (0 for market).
        order_type: 'market', 'limit', or 'stop'.
        asset_class: Asset class for routing.
        preferred_exchange: Override automatic routing.

    Returns:
        RoutedOrder with exchange assignment and status.
    """
    self._order_counter += 1
    order_id = f"RT-{self._order_counter:08d}"

    # Find target exchange
    target = None
    if preferred_exchange:
        target = self._exchanges.get(preferred_exchange)
        if target and not target.is_active:
            target = None

    if target is None:
        candidates = self.find_exchanges(asset_class)
        target = candidates[0] if candidates else None

    if target is None:
        order = RoutedOrder(
            order_id=order_id, symbol=symbol, side=side,
            quantity=quantity, price=price, order_type=order_type,
            status="rejected", reject_reason=f"No active exchange for {asset_class}",
        )
        self._order_history.append(order)
        logger.warning(f"Order {order_id} rejected: no exchange for {asset_class}")
        return order

    order = RoutedOrder(
        order_id=order_id, symbol=symbol, side=side,
        quantity=quantity, price=price, order_type=order_type,
        exchange=target.name, status="routed",
    )

    # Execute if handler available
    if target.execute_fn:
        try:
            result = target.execute_fn(order)
            order.status = "filled"
            order.fill_price = result.get("price", price)
            order.fill_quantity = result.get("quantity", quantity)
        except Exception as e:
            order.status = "rejected"
            order.reject_reason = str(e)
    else:
        order.status = "routed"

    self._order_history.append(order)
    logger.info(f"Order {order_id}: {side} {quantity} {symbol} → {target.name} [{order.status}]")
    return order

financelib.enterprise.order_router.Exchange(name: str, asset_classes: List[str] = list(), priority: int = 0, status: ExchangeStatus = ExchangeStatus.ACTIVE, execute_fn: Optional[Callable] = None, metadata: Dict[str, Any] = dict()) dataclass

Represents a trading exchange/venue.

Parameters:

Name Type Description Default
name str

Exchange identifier (e.g., 'BIST', 'BINANCE', 'NYSE').

 required
asset_classes List[str]

Supported asset classes.

 list()
priority int

Routing priority (lower = higher priority).

 0
status ExchangeStatus

Current exchange status.

 ACTIVE
execute_fn Optional[Callable]

Optional function to execute orders.

 None

financelib.enterprise.order_router.RoutedOrder(order_id: str = '', symbol: str = '', side: str = '', quantity: float = 0.0, price: float = 0.0, order_type: str = 'market', exchange: str = '', status: str = 'pending', timestamp: str = (lambda: datetime.now().isoformat())(), fill_price: float = 0.0, fill_quantity: float = 0.0, reject_reason: str = '') dataclass

An order routed to a specific exchange.

FIX-like message format for order representation.

Attributes

fix_msg: Dict[str, Any] property

Generate FIX-like message representation.

Audit Trail

financelib.enterprise.audit.AuditTrail()

Thread-safe audit trail for all trading operations.

Records every action with timestamp, user, and details. Entries are immutable once created.

Example

audit = AuditTrail() audit.log("ORDER_PLACED", "alice", {"symbol": "THYAO.IS", "side": "buy"}) audit.log("ORDER_FILLED", "alice", {"symbol": "THYAO.IS", "price": 175.0}) print(audit.query(user_id="alice"))

Source code in financelib/enterprise/audit.py 
57
58
59
60
def __init__(self) -> None:
    self._entries: List[AuditEntry] = []
    self._counter = 0
    self._lock = threading.Lock()

Functions

clear() -> None

Clear all entries (use with caution — breaks audit integrity).

Source code in financelib/enterprise/audit.py 
135
136
137
138
139
def clear(self) -> None:
    """Clear all entries (use with caution — breaks audit integrity)."""
    with self._lock:
        self._entries.clear()
        self._counter = 0

export_json() -> str

Export all entries as JSON array.

Source code in financelib/enterprise/audit.py 
114
115
116
def export_json(self) -> str:
    """Export all entries as JSON array."""
    return json.dumps([e.to_dict() for e in self._entries], default=str, indent=2)

export_to_file(filepath: str) -> int

Export audit trail to a JSON file.

Parameters:

Name Type Description Default
filepath str

Output file path.

 required

Returns:

Type Description
int

Number of entries exported.
Source code in financelib/enterprise/audit.py 
118
119
120
121
122
123
124
125
126
127
128
129
def export_to_file(self, filepath: str) -> int:
    """Export audit trail to a JSON file.

    Args:
        filepath: Output file path.

    Returns:
        Number of entries exported.
    """
    with open(filepath, "w") as f:
        f.write(self.export_json())
    return len(self._entries)

log(action: str, user_id: str = '', details: Optional[Dict] = None) -> AuditEntry

Record an audit entry.

Parameters:

Name Type Description Default
action str

Action type string.

 required
user_id str

User identifier.

 ''
details Optional[Dict]

Action-specific data.

 None

Returns:

Type Description
AuditEntry

The created AuditEntry.
Source code in financelib/enterprise/audit.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def log(self, action: str, user_id: str = "", details: Optional[Dict] = None) -> AuditEntry:
    """Record an audit entry.

    Args:
        action: Action type string.
        user_id: User identifier.
        details: Action-specific data.

    Returns:
        The created AuditEntry.
    """
    with self._lock:
        self._counter += 1
        entry = AuditEntry(
            timestamp=datetime.now().isoformat(),
            action=action,
            user_id=user_id,
            details=details or {},
            entry_id=self._counter,
        )
        self._entries.append(entry)
    return entry

query(action: Optional[str] = None, user_id: Optional[str] = None, since: Optional[str] = None, limit: int = 100) -> List[AuditEntry]

Query audit entries with filters.

Parameters:

Name Type Description Default
action Optional[str]

Filter by action type.

 None
user_id Optional[str]

Filter by user.

 None
since Optional[str]

Filter entries after this ISO timestamp.

 None
limit int

Maximum entries to return.

 100

Returns:

Type Description
List[AuditEntry]

Matching audit entries (newest first).
Source code in financelib/enterprise/audit.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def query(
    self,
    action: Optional[str] = None,
    user_id: Optional[str] = None,
    since: Optional[str] = None,
    limit: int = 100,
) -> List[AuditEntry]:
    """Query audit entries with filters.

    Args:
        action: Filter by action type.
        user_id: Filter by user.
        since: Filter entries after this ISO timestamp.
        limit: Maximum entries to return.

    Returns:
        Matching audit entries (newest first).
    """
    results = self._entries.copy()

    if action:
        results = [e for e in results if e.action == action]
    if user_id:
        results = [e for e in results if e.user_id == user_id]
    if since:
        results = [e for e in results if e.timestamp >= since]

    return list(reversed(results[-limit:]))

financelib.enterprise.audit.AuditEntry(timestamp: str, action: str, user_id: str = '', details: Dict[str, Any] = dict(), entry_id: int = 0) dataclass

Immutable audit log entry.

Attributes:

Name Type Description
timestamp str

ISO-format timestamp.
action str

Action type (e.g., 'ORDER_PLACED', 'POSITION_OPENED').
user_id str

User who performed the action.
details Dict[str, Any]

Action-specific data.
entry_id int

Unique entry identifier.

Compliance

financelib.enterprise.compliance.TaxReport(year: int = 2026, currency: str = 'TRY')

Capital gains tax report generator.

Calculates realized gains/losses using FIFO method and generates tax-ready reports.

Parameters:

Name Type Description Default
year int

Tax year.

 2026
currency str

Reporting currency.

 'TRY'
Example

report = TaxReport(year=2026) report.add_trade(TradeForReport(symbol="THYAO.IS", side="buy", quantity=100, price=170)) report.add_trade(TradeForReport(symbol="THYAO.IS", side="sell", quantity=100, price=180)) summary = report.generate() print(f"Net gain: {summary['net_gain']}")

Source code in financelib/enterprise/compliance.py 
51
52
53
54
def __init__(self, year: int = 2026, currency: str = "TRY") -> None:
    self.year = year
    self.currency = currency
    self._trades: List[TradeForReport] = []

Functions

generate() -> Dict[str, Any]

Generate capital gains tax report using FIFO.

Returns:

Type Description
Dict[str, Any]

Dictionary with total gains, losses, net, and per-symbol breakdown.
Source code in financelib/enterprise/compliance.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def generate(self) -> Dict[str, Any]:
    """Generate capital gains tax report using FIFO.

    Returns:
        Dictionary with total gains, losses, net, and per-symbol breakdown.
    """
    # Group by symbol
    buys: Dict[str, List[TradeForReport]] = {}
    sells: Dict[str, List[TradeForReport]] = {}

    for t in self._trades:
        if t.side == "buy":
            buys.setdefault(t.symbol, []).append(t)
        elif t.side == "sell":
            sells.setdefault(t.symbol, []).append(t)

    total_gains = 0.0
    total_losses = 0.0
    total_commission = sum(t.commission for t in self._trades)
    symbol_breakdown: List[Dict[str, Any]] = []

    for symbol, sell_list in sells.items():
        buy_queue = list(buys.get(symbol, []))
        symbol_gain = 0.0

        for sell in sell_list:
            remaining = sell.quantity
            while remaining > 0 and buy_queue:
                buy = buy_queue[0]
                matched = min(remaining, buy.quantity)
                gain = (sell.price - buy.price) * matched
                symbol_gain += gain
                remaining -= matched
                buy.quantity -= matched
                if buy.quantity <= 0:
                    buy_queue.pop(0)

        if symbol_gain >= 0:
            total_gains += symbol_gain
        else:
            total_losses += abs(symbol_gain)

        symbol_breakdown.append({
            "symbol": symbol,
            "gain_loss": round(symbol_gain, 2),
            "num_trades": len(sell_list),
        })

    net = total_gains - total_losses - total_commission

    return {
        "year": self.year,
        "currency": self.currency,
        "total_gains": round(total_gains, 2),
        "total_losses": round(total_losses, 2),
        "total_commission": round(total_commission, 2),
        "net_gain": round(net, 2),
        "num_trades": len(self._trades),
        "symbols": symbol_breakdown,
    }

financelib.enterprise.compliance.MiFIDReport(firm_id: str = '', lei: str = '')

MiFID II transaction reporting.

Generates regulatory transaction reports compliant with MiFID II/MiFIR requirements (simplified).

Example

mifid = MiFIDReport(firm_id="FINLIB001") mifid.add_trade(trade) report = mifid.generate()

Source code in financelib/enterprise/compliance.py 
136
137
138
139
def __init__(self, firm_id: str = "", lei: str = "") -> None:
    self.firm_id = firm_id
    self.lei = lei  # Legal Entity Identifier
    self._trades: List[TradeForReport] = []

Functions

generate() -> List[Dict[str, Any]]

Generate MiFID II transaction reports.

Returns:

Type Description
List[Dict[str, Any]]

List of transaction report records in MiFID II format.
Source code in financelib/enterprise/compliance.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def generate(self) -> List[Dict[str, Any]]:
    """Generate MiFID II transaction reports.

    Returns:
        List of transaction report records in MiFID II format.
    """
    reports = []
    for t in self._trades:
        reports.append({
            # RTS 25 fields (simplified)
            "report_status": "NEW",
            "trading_date_time": t.timestamp or datetime.now().isoformat(),
            "trading_capacity": "DEAL",  # Dealing on own account
            "instrument_id": t.symbol,
            "price": t.price,
            "quantity": t.quantity,
            "buy_sell_indicator": "BUYI" if t.side == "buy" else "SELL",
            "venue": t.exchange or "XIST",  # BIST MIC code
            "currency": t.currency,
            "investment_firm_id": self.firm_id,
            "lei": self.lei,
            "counterparty": t.counterparty,
            "commission": t.commission,
            "transaction_ref": t.trade_id,
        })

    logger.info(f"Generated {len(reports)} MiFID II transaction reports")
    return reports

financelib.enterprise.compliance.TradeForReport(trade_id: str = '', symbol: str = '', side: str = '', quantity: float = 0.0, price: float = 0.0, commission: float = 0.0, exchange: str = '', timestamp: str = '', currency: str = 'TRY', counterparty: str = '') dataclass

Standardized trade record for compliance reporting.

RBAC

financelib.enterprise.rbac.RBACManager()

Role-Based Access Control manager.

Manages user-role assignments and permission checks.

Example

rbac = RBACManager() rbac.assign_role("alice", "trader") rbac.assign_role("bob", "viewer") assert rbac.check_permission("alice", Permission.TRADE) is True assert rbac.check_permission("bob", Permission.TRADE) is False

Source code in financelib/enterprise/rbac.py 
69
70
71
72
73
74
75
76
77
def __init__(self) -> None:
    self._roles: Dict[str, Role] = {
        "viewer": ROLE_VIEWER,
        "trader": ROLE_TRADER,
        "manager": ROLE_MANAGER,
        "admin": ROLE_ADMIN,
    }
    self._user_roles: Dict[str, str] = {}
    self._lock = threading.Lock()

Functions

add_role(role: Role) -> None

Register a custom role.

Source code in financelib/enterprise/rbac.py 
79
80
81
82
def add_role(self, role: Role) -> None:
    """Register a custom role."""
    with self._lock:
        self._roles[role.name] = role

assign_role(user_id: str, role_name: str) -> bool

Assign a role to a user.

Parameters:

Name Type Description Default
user_id str

User identifier.

 required
role_name str

Role name to assign.

 required

Returns:

Type Description
bool

True if assigned, False if role doesn't exist.
Source code in financelib/enterprise/rbac.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def assign_role(self, user_id: str, role_name: str) -> bool:
    """Assign a role to a user.

    Args:
        user_id: User identifier.
        role_name: Role name to assign.

    Returns:
        True if assigned, False if role doesn't exist.
    """
    if role_name not in self._roles:
        logger.error(f"Role '{role_name}' not found")
        return False
    with self._lock:
        self._user_roles[user_id] = role_name
    logger.info(f"Role '{role_name}' assigned to '{user_id}'")
    return True

check_permission(user_id: str, permission: Permission) -> bool

Check if a user has a specific permission.

Parameters:

Name Type Description Default
user_id str

User identifier.

 required
permission Permission

Permission to check.

 required

Returns:

Type Description
bool

True if allowed.
Source code in financelib/enterprise/rbac.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def check_permission(self, user_id: str, permission: Permission) -> bool:
    """Check if a user has a specific permission.

    Args:
        user_id: User identifier.
        permission: Permission to check.

    Returns:
        True if allowed.
    """
    role = self.get_user_role(user_id)
    if role is None:
        return False
    return role.has_permission(permission)

get_user_role(user_id: str) -> Optional[Role]

Get a user's role object.

Source code in financelib/enterprise/rbac.py 
110
111
112
113
114
115
def get_user_role(self, user_id: str) -> Optional[Role]:
    """Get a user's role object."""
    role_name = self._user_roles.get(user_id)
    if role_name:
        return self._roles.get(role_name)
    return None

get_users_with_role(role_name: str) -> List[str]

Get all users with a specific role.

Source code in financelib/enterprise/rbac.py 
132
133
134
def get_users_with_role(self, role_name: str) -> List[str]:
    """Get all users with a specific role."""
    return [uid for uid, r in self._user_roles.items() if r == role_name]

list_roles() -> List[str]

List all available role names.

Source code in financelib/enterprise/rbac.py 
136
137
138
def list_roles(self) -> List[str]:
    """List all available role names."""
    return list(self._roles.keys())

list_users() -> Dict[str, str]

List all user-role assignments.

Source code in financelib/enterprise/rbac.py 
140
141
142
def list_users(self) -> Dict[str, str]:
    """List all user-role assignments."""
    return dict(self._user_roles)

revoke_role(user_id: str) -> bool

Remove a user's role.

Source code in financelib/enterprise/rbac.py 
102
103
104
105
106
107
108
def revoke_role(self, user_id: str) -> bool:
    """Remove a user's role."""
    with self._lock:
        if user_id in self._user_roles:
            del self._user_roles[user_id]
            return True
    return False

financelib.enterprise.rbac.Role(name: str, permissions: Set[Permission] = set()) dataclass

A named role with a set of permissions.

Example

trader = Role("trader", {Permission.VIEW_PORTFOLIO, Permission.TRADE}) admin = Role("admin", {p for p in Permission})

financelib.enterprise.rbac.Permission

Bases: Enum

Available permissions.

Plugin Architecture

financelib.enterprise.plugin.PluginManager()

Manages plugin lifecycle — registration, initialization, and discovery.

Example

manager = PluginManager() manager.register(MyDataPlugin) manager.initialize_all() plugin = manager.get("my_data") manager.shutdown_all()

Source code in financelib/enterprise/plugin.py 
70
71
72
def __init__(self) -> None:
    self._plugins: Dict[str, BasePlugin] = {}
    self._registered: Dict[str, Type[BasePlugin]] = {}

Functions

get(name: str) -> Optional[BasePlugin]

Get a plugin instance by name.

Source code in financelib/enterprise/plugin.py 
87
88
89
def get(self, name: str) -> Optional[BasePlugin]:
    """Get a plugin instance by name."""
    return self._plugins.get(name)

health_check_all() -> Dict[str, bool]

Run health checks on all plugins.

Source code in financelib/enterprise/plugin.py 
121
122
123
def health_check_all(self) -> Dict[str, bool]:
    """Run health checks on all plugins."""
    return {name: p.health_check() for name, p in self._plugins.items()}

initialize_all(config: Optional[Dict[str, Dict]] = None) -> None

Initialize all registered plugins.

Parameters:

Name Type Description Default
config Optional[Dict[str, Dict]]

Dict mapping plugin name → config dict.

 None
Source code in financelib/enterprise/plugin.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def initialize_all(self, config: Optional[Dict[str, Dict]] = None) -> None:
    """Initialize all registered plugins.

    Args:
        config: Dict mapping plugin name → config dict.
    """
    config = config or {}
    for name, plugin in self._plugins.items():
        try:
            plugin.initialize(config.get(name))
            logger.info(f"Plugin initialized: {name}")
        except Exception as e:
            logger.error(f"Plugin init failed: {name} — {e}")

list_plugins() -> List[Dict[str, str]]

List all registered plugins with name and version.

Source code in financelib/enterprise/plugin.py 
114
115
116
117
118
119
def list_plugins(self) -> List[Dict[str, str]]:
    """List all registered plugins with name and version."""
    return [
        {"name": p.name, "version": p.version, "healthy": p.health_check()}
        for p in self._plugins.values()
    ]

register(plugin_class: Type[BasePlugin], config: Optional[Dict] = None) -> None

Register and instantiate a plugin.

Parameters:

Name Type Description Default
plugin_class Type[BasePlugin]

Plugin class (not instance).

 required
config Optional[Dict]

Optional configuration dict.

 None
Source code in financelib/enterprise/plugin.py 
74
75
76
77
78
79
80
81
82
83
84
85
def register(self, plugin_class: Type[BasePlugin], config: Optional[Dict] = None) -> None:
    """Register and instantiate a plugin.

    Args:
        plugin_class: Plugin class (not instance).
        config: Optional configuration dict.
    """
    instance = plugin_class()
    name = instance.name
    self._registered[name] = plugin_class
    self._plugins[name] = instance
    logger.info(f"Plugin registered: {name} v{instance.version}")

shutdown_all() -> None

Shutdown all plugins gracefully.

Source code in financelib/enterprise/plugin.py 
105
106
107
108
109
110
111
112
def shutdown_all(self) -> None:
    """Shutdown all plugins gracefully."""
    for name, plugin in self._plugins.items():
        try:
            plugin.shutdown()
            logger.info(f"Plugin shutdown: {name}")
        except Exception as e:
            logger.error(f"Plugin shutdown failed: {name} — {e}")

unregister(name: str) -> bool

Remove a plugin.

Source code in financelib/enterprise/plugin.py 
129
130
131
132
133
134
135
136
137
138
139
def unregister(self, name: str) -> bool:
    """Remove a plugin."""
    if name in self._plugins:
        try:
            self._plugins[name].shutdown()
        except Exception:
            pass
        del self._plugins[name]
        del self._registered[name]
        return True
    return False

financelib.enterprise.plugin.BasePlugin

Bases: ABC

Base class for all FinanceLib plugins.

Plugins must implement name, version, initialize, and shutdown.

Example

class MyDataPlugin(BasePlugin): ... @property ... def name(self): return "my_data" ... @property ... def version(self): return "1.0" ... def initialize(self, config): print("Started!") ... def shutdown(self): print("Stopped!")

Attributes

name: str abstractmethod property

Plugin unique name.

version: str abstractmethod property

Plugin version string.

Functions

health_check() -> bool

Optional health check. Override to implement.

Source code in financelib/enterprise/plugin.py 
54
55
56
def health_check(self) -> bool:
    """Optional health check. Override to implement."""
    return True

initialize(config: Optional[Dict[str, Any]] = None) -> None abstractmethod

Initialize the plugin with optional configuration.

Source code in financelib/enterprise/plugin.py 
44
45
46
47
@abstractmethod
def initialize(self, config: Optional[Dict[str, Any]] = None) -> None:
    """Initialize the plugin with optional configuration."""
    ...

shutdown() -> None abstractmethod

Clean shutdown of the plugin.

Source code in financelib/enterprise/plugin.py 
49
50
51
52
@abstractmethod
def shutdown(self) -> None:
    """Clean shutdown of the plugin."""
    ...