investfly.models.strategy.StrategyServices

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:

Construct the common protective-exit DTO; the runtime owns its evaluation.

@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.

CustomStrategyPolicy( assetType: investfly.models.marketdata.SecurityType, positionManagement: Optional[investfly.models.strategy.PositionManagementRules] = None, assetLifecycle: Optional[investfly.models.strategy.StrategyConfig.AssetLifecycleRules] = None, portfolioLimits: Optional[investfly.models.strategy.PortfolioLimits] = None)
positionManagement: Optional[investfly.models.strategy.PositionManagementRules] = None
portfolioLimits: Optional[investfly.models.strategy.PortfolioLimits] = None
def validate(self) -> None:
def protectiveExitPlan( self) -> investfly.models.strategy.ProtectiveExitPlan | None:
@dataclass(frozen=True)
class RankedSecurity:
RankedSecurity( security: investfly.models.marketdata.Security, score: float)
score: float
@dataclass
class OpenOrderRequest:

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

OpenOrderRequest( security: investfly.models.marketdata.Security, execution: investfly.models.strategy.OpenExecutionSettings, side: investfly.models.portfolio.PositionType = LONG, allowAddToExisting: bool = False, existingTargetPolicy: Optional[investfly.models.strategy.ExistingTargetPolicy] = None)
allowAddToExisting: bool = False
existingTargetPolicy: Optional[investfly.models.strategy.ExistingTargetPolicy] = None
@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.

AllocationPlan( securities: List[investfly.models.marketdata.Security], execution: investfly.models.strategy.OpenExecutionSettings, side: investfly.models.portfolio.PositionType = LONG, allowAddToExisting: bool = False)
allowAddToExisting: bool = False
class TargetWeightMode(builtins.str, enum.Enum):

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

EQUAL_WEIGHT = <TargetWeightMode.EQUAL_WEIGHT: 'EQUAL_WEIGHT'>
SCORE_WEIGHTED = <TargetWeightMode.SCORE_WEIGHTED: 'SCORE_WEIGHTED'>
class ReplacementPolicy(builtins.str, enum.Enum):

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

REPLACE_DROPPED = <ReplacementPolicy.REPLACE_DROPPED: 'REPLACE_DROPPED'>
ADD_ONLY = <ReplacementPolicy.ADD_ONLY: 'ADD_ONLY'>
@dataclass
class RebalancePlan:
RebalancePlan( targetSecurities: investfly.models.strategy.RankedSecuritySelection, allocationBudget: investfly.models.strategy.PositionSizeSpec, targetWeightMode: TargetWeightMode = <TargetWeightMode.EQUAL_WEIGHT: 'EQUAL_WEIGHT'>, replacementPolicy: ReplacementPolicy = <ReplacementPolicy.REPLACE_DROPPED: 'REPLACE_DROPPED'>, rebalanceTolerance: Optional[float] = 0.05, targetWeights: Optional[List[investfly.models.strategy.TargetWeightSpec]] = None)
targetWeightMode: TargetWeightMode = <TargetWeightMode.EQUAL_WEIGHT: 'EQUAL_WEIGHT'>
replacementPolicy: ReplacementPolicy = <ReplacementPolicy.REPLACE_DROPPED: 'REPLACE_DROPPED'>
rebalanceTolerance: Optional[float] = 0.05
targetWeights: Optional[List[investfly.models.strategy.TargetWeightSpec]] = None
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 rankSecurities( self, selection: investfly.models.strategy.RankedSecuritySelection) -> List[RankedSecurity]:
@abstractmethod
def evaluateCondition( self, condition: investfly.models.strategy.SecurityFilterExpression, security: investfly.models.marketdata.Security) -> bool:
@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]:
@abstractmethod
def planOpenOrders( self, request: OpenOrderRequest) -> List[investfly.models.portfolio.TradeOrder]:
@abstractmethod
def planAllocationOrders( self, plan: AllocationPlan) -> List[investfly.models.portfolio.TradeOrder]:
@abstractmethod
def planCloseOrders( self, positions: List[investfly.models.portfolio.OpenPosition], closeSpec: investfly.models.strategy.ClosePositionSpec | None = None) -> List[investfly.models.portfolio.TradeOrder]:
@abstractmethod
def planRebalanceOrders( self, plan: RebalancePlan) -> List[investfly.models.portfolio.TradeOrder]:
@abstractmethod
def planScalingOrders( self, updatedSecurities: List[investfly.models.marketdata.Security]) -> List[investfly.models.portfolio.TradeOrder]:
@abstractmethod
def drainWarnings(self) -> List[str]: