investfly.models.strategy.StrategyServices

Runtime planning and evaluation services available to custom strategies.

def buildProtectiveExitPlan( targetProfitPct: float | None = None, stopLossPct: float | None = None, trailingStopPct: float | None = None, maxHold: investfly.models.strategy.StrategyDuration | None = None) -> investfly.models.strategy.ProtectiveExitPlan:

Build a protective-exit plan from common percentage-based settings.

The returned plan may combine a profit target, fixed stop, trailing stop, and maximum holding period. The Investfly runtime evaluates the plan.

@dataclass
class CustomStrategyPolicy:

Continuous behavior that the runtime manages around custom strategy callbacks.

A custom strategy remains responsible for deciding when to request an entry or rebalance. The runtime owns the configured position management, asset lifecycle, and portfolio-limit algorithms before and after those callbacks.

@dataclass(frozen=True)
class RankedSecurity:

A selected security and its ranking score.

@dataclass
class OpenOrderRequest:

Request runtime-managed instrument selection, sizing, and order construction.

@dataclass
class AllocationPlan:

Allocate one total execution budget equally across the supplied securities.

For percentage and notional sizing, execution.positionSize is the total pool and the runtime divides it by the number of securities. For fixed quantity sizing, the quantity is applied to each security.

class TargetWeightMode(builtins.str, enum.Enum):

How a rebalance budget is distributed among selected securities.

class ReplacementPolicy(builtins.str, enum.Enum):

How a rebalance handles securities that leave the target set.

@dataclass
class RebalancePlan:

Describe target selection, weighting, and replacement for a rebalance.

class StrategyServices(abc.ABC):

Runtime-injected operations shared with configuration-driven strategies.

This class is a public contract only. Strategy authors call self.services; Investfly injects the live/backtest implementation before any strategy callback is evaluated.

@abstractmethod
def selectSecurities( self, selection: investfly.models.strategy.SecuritySelection | investfly.models.strategy.RankedSecuritySelection) -> List[investfly.models.marketdata.Security]:

Resolve a selection specification to ordered securities.

@abstractmethod
def rankSecurities( self, selection: investfly.models.strategy.RankedSecuritySelection) -> List[RankedSecurity]:

Resolve and score securities using a ranked selection specification.

@abstractmethod
def evaluateCondition( self, condition: investfly.models.strategy.SecurityFilterExpression, security: investfly.models.marketdata.Security) -> bool:

Evaluate a filter expression for one security.

@abstractmethod
def evaluateGuards( self, policy: investfly.models.strategy.GuardPolicy, scope: investfly.models.strategy.GuardScope = <GuardScope.SCHEDULED_JOB: 'SCHEDULED_JOB'>, targetKey: str | None = None) -> List[investfly.models.strategy.GuardDecision]:

Evaluate strategy guards and return their decisions for the requested scope.

@abstractmethod
def planOpenOrders( self, request: OpenOrderRequest) -> List[investfly.models.portfolio.TradeOrder]:

Plan entry orders with runtime-managed selection, sizing, and execution.

@abstractmethod
def planAllocationOrders( self, plan: AllocationPlan) -> List[investfly.models.portfolio.TradeOrder]:

Plan orders that allocate a budget across a group of securities.

@abstractmethod
def planCloseOrders( self, positions: List[investfly.models.portfolio.OpenPosition], closeSpec: investfly.models.strategy.ClosePositionSpec | None = None) -> List[investfly.models.portfolio.TradeOrder]:

Plan closing orders for the supplied open positions.

@abstractmethod
def planRebalanceOrders( self, plan: RebalancePlan) -> List[investfly.models.portfolio.TradeOrder]:

Plan orders that move the portfolio toward a target basket.

@abstractmethod
def planScalingOrders( self, updatedSecurities: List[investfly.models.marketdata.Security]) -> List[investfly.models.portfolio.TradeOrder]:

Plan scale-in orders for positions affected by the latest market update.

@abstractmethod
def drainWarnings(self) -> List[str]:

Return and clear non-fatal warnings produced by previous service calls.