investfly.models.strategy.StrategyConfig

class EdgeMode(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'.

RISING_EDGE = <EdgeMode.RISING_EDGE: 'RISING_EDGE'>
WHEN_TRUE = <EdgeMode.WHEN_TRUE: 'WHEN_TRUE'>
class EntryConflictAction(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'.

IGNORE = <EntryConflictAction.IGNORE: 'IGNORE'>
CLOSE_OPPOSITE_THEN_OPEN = <EntryConflictAction.CLOSE_OPPOSITE_THEN_OPEN: 'CLOSE_OPPOSITE_THEN_OPEN'>
OPEN_HEDGED_POSITION = <EntryConflictAction.OPEN_HEDGED_POSITION: 'OPEN_HEDGED_POSITION'>
@dataclass
class EntryConflictPolicy:
EntryConflictPolicy( action: Optional[EntryConflictAction] = <EntryConflictAction.IGNORE: 'IGNORE'>, allowHedging: Optional[bool] = False)
action: Optional[EntryConflictAction] = <EntryConflictAction.IGNORE: 'IGNORE'>
allowHedging: Optional[bool] = False
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType]) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[EntryConflictPolicy]:
@dataclass
class EntryExistingPositionPolicy:
EntryExistingPositionPolicy( sameTarget: Optional[ExistingTargetPolicy] = None, oppositeSide: Optional[EntryConflictPolicy] = None)
sameTarget: Optional[ExistingTargetPolicy] = None
oppositeSide: Optional[EntryConflictPolicy] = None
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType]) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[EntryExistingPositionPolicy]:
class ExistingTargetAction(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'.

IGNORE_NEW_ENTRY = <ExistingTargetAction.IGNORE_NEW_ENTRY: 'IGNORE_NEW_ENTRY'>
ADD_TO_EXISTING = <ExistingTargetAction.ADD_TO_EXISTING: 'ADD_TO_EXISTING'>
REPLACE_EXISTING = <ExistingTargetAction.REPLACE_EXISTING: 'REPLACE_EXISTING'>
@dataclass
class ExistingTargetPolicy:
ExistingTargetPolicy( action: ExistingTargetAction = <ExistingTargetAction.IGNORE_NEW_ENTRY: 'IGNORE_NEW_ENTRY'>, addRule: Optional[RepeatSignalAddRule] = None)
addRule: Optional[RepeatSignalAddRule] = None
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType]) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[ExistingTargetPolicy]:
@dataclass
class EntryRule:
EntryRule( edgeMode: EdgeMode = <EdgeMode.RISING_EDGE: 'RISING_EDGE'>, condition: investfly.models.strategy.SecurityFilterExpression = None, runWhen: Optional[GuardPolicy] = None, existingPositionPolicy: Optional[EntryExistingPositionPolicy] = <factory>)
edgeMode: EdgeMode = <EdgeMode.RISING_EDGE: 'RISING_EDGE'>
runWhen: Optional[GuardPolicy] = None
existingPositionPolicy: Optional[EntryExistingPositionPolicy]
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> EntryRule:
@dataclass
class OpenExecutionSettings:
OpenExecutionSettings( positionSize: Optional[PositionSizeSpec] = None, orderSpec: Optional[OrderSpec] = None, instrumentSelection: Optional[investfly.models.strategy.InstrumentSelectionSpec] = None)
positionSize: Optional[PositionSizeSpec] = None
orderSpec: Optional[OrderSpec] = None
instrumentSelection: Optional[investfly.models.strategy.InstrumentSelectionSpec] = None
futureContractSelector: Optional[investfly.models.strategy.FutureContractSelector]
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[OpenExecutionSettings]:
@dataclass
class EntryRules:
EntryRules( universe: Optional[investfly.models.strategy.SecurityUniverseSelector] = None, execution: Optional[OpenExecutionSettings] = None, longEntry: Optional[EntryRule] = None, shortEntry: Optional[EntryRule] = None, optionEntry: Optional[EntryRule] = None)
execution: Optional[OpenExecutionSettings] = None
longEntry: Optional[EntryRule] = None
shortEntry: Optional[EntryRule] = None
optionEntry: Optional[EntryRule] = None
def positionEntries(self) -> List[EntryRule]:
def allEntries(self) -> List[EntryRule]:
def hasConfiguredEntry(self) -> bool:
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> EntryRules:
@dataclass
class ExitRule:
ExitRule( edgeMode: EdgeMode = <EdgeMode.WHEN_TRUE: 'WHEN_TRUE'>, condition: investfly.models.strategy.SecurityFilterExpression = None, closePosition: Optional[ClosePositionSpec] = None)
edgeMode: EdgeMode = <EdgeMode.WHEN_TRUE: 'WHEN_TRUE'>
closePosition: Optional[ClosePositionSpec] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ExitRule:
@dataclass
class ExitRules:
ExitRules( signalExit: Optional[ExitRule] = None, protectiveExits: Optional[ProtectiveExitPlan] = None)
signalExit: Optional[ExitRule] = None
protectiveExits: Optional[ProtectiveExitPlan] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> ExitRules:
class ProfitThresholdType(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'.

PERCENT_GAIN = <ProfitThresholdType.PERCENT_GAIN: 'PERCENT_GAIN'>
ABSOLUTE_PNL = <ProfitThresholdType.ABSOLUTE_PNL: 'ABSOLUTE_PNL'>
PRICE = <ProfitThresholdType.PRICE: 'PRICE'>
TICKS = <ProfitThresholdType.TICKS: 'TICKS'>
POINTS = <ProfitThresholdType.POINTS: 'POINTS'>
PIPS = <ProfitThresholdType.PIPS: 'PIPS'>
R_MULTIPLE = <ProfitThresholdType.R_MULTIPLE: 'R_MULTIPLE'>
@dataclass
class ProfitTargetTier:
ProfitTargetTier(threshold: float, closePercent: Optional[float] = None)
threshold: float
closePercent: Optional[float] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ProfitTargetTier:
@dataclass
class StagedProfitTargets:
StagedProfitTargets( type: ProfitThresholdType, tiers: List[ProfitTargetTier] = <factory>)
tiers: List[ProfitTargetTier]
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def percentGain( threshold: float, closePercent: float) -> StagedProfitTargets:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> StagedProfitTargets:
class FixedStopThresholdType(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'.

PERCENT_FROM_ENTRY = <FixedStopThresholdType.PERCENT_FROM_ENTRY: 'PERCENT_FROM_ENTRY'>
ABSOLUTE_PNL = <FixedStopThresholdType.ABSOLUTE_PNL: 'ABSOLUTE_PNL'>
PRICE = <FixedStopThresholdType.PRICE: 'PRICE'>
TICKS_FROM_ENTRY = <FixedStopThresholdType.TICKS_FROM_ENTRY: 'TICKS_FROM_ENTRY'>
POINTS_FROM_ENTRY = <FixedStopThresholdType.POINTS_FROM_ENTRY: 'POINTS_FROM_ENTRY'>
PIPS_FROM_ENTRY = <FixedStopThresholdType.PIPS_FROM_ENTRY: 'PIPS_FROM_ENTRY'>
ATR_MULTIPLE_FROM_ENTRY = <FixedStopThresholdType.ATR_MULTIPLE_FROM_ENTRY: 'ATR_MULTIPLE_FROM_ENTRY'>
@dataclass
class FixedStopThreshold:
FixedStopThreshold( type: FixedStopThresholdType, value: float, atrParams: Optional[AtrParams] = None)
value: float
atrParams: Optional[AtrParams] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None, replacement: bool = False) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> FixedStopThreshold:
@dataclass
class FixedStopRule:
FixedStopRule( threshold: FixedStopThreshold)
threshold: FixedStopThreshold
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None, replacement: bool = False) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> FixedStopRule:
class TrailingStopDistanceType(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'.

PERCENT = <TrailingStopDistanceType.PERCENT: 'PERCENT'>
ATR_MULTIPLE = <TrailingStopDistanceType.ATR_MULTIPLE: 'ATR_MULTIPLE'>
PRICE = <TrailingStopDistanceType.PRICE: 'PRICE'>
TICKS = <TrailingStopDistanceType.TICKS: 'TICKS'>
POINTS = <TrailingStopDistanceType.POINTS: 'POINTS'>
@dataclass
class AtrParams:
AtrParams( period: int = 14, barInterval: investfly.models.marketdata.BarInterval = <BarInterval.ONE_DAY: 'ONE_DAY'>)
period: int = 14
barInterval: investfly.models.marketdata.BarInterval = <BarInterval.ONE_DAY: 'ONE_DAY'>
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> AtrParams:
@dataclass
class TrailingStopDistance:
TrailingStopDistance( type: TrailingStopDistanceType, value: float, atrParams: Optional[AtrParams] = None)
value: float
atrParams: Optional[AtrParams] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> TrailingStopDistance:
@dataclass
class TrailingStopRule:
TrailingStopRule( distance: TrailingStopDistance)
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> TrailingStopRule:
class SessionTimeTriggerType(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'.

AT_TIME = <SessionTimeTriggerType.AT_TIME: 'AT_TIME'>
BEFORE_SESSION_CLOSE = <SessionTimeTriggerType.BEFORE_SESSION_CLOSE: 'BEFORE_SESSION_CLOSE'>
@dataclass
class SessionTimeTrigger:
SessionTimeTrigger( type: SessionTimeTriggerType = <SessionTimeTriggerType.BEFORE_SESSION_CLOSE: 'BEFORE_SESSION_CLOSE'>, time: Optional[str] = None, offsetBeforeClose: Optional[investfly.models.common.TimeDelta] = None, timeZonePolicy: Optional[str] = 'MARKET')
time: Optional[str] = None
offsetBeforeClose: Optional[investfly.models.common.TimeDelta] = None
timeZonePolicy: Optional[str] = 'MARKET'
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> SessionTimeTrigger:
class ProtectiveExecutionPolicy(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'.

BEST_AVAILABLE = <ProtectiveExecutionPolicy.BEST_AVAILABLE: 'BEST_AVAILABLE'>
BROKER_NATIVE_REQUIRED = <ProtectiveExecutionPolicy.BROKER_NATIVE_REQUIRED: 'BROKER_NATIVE_REQUIRED'>
INVESTFLY_MANAGED_ONLY = <ProtectiveExecutionPolicy.INVESTFLY_MANAGED_ONLY: 'INVESTFLY_MANAGED_ONLY'>
class StopSpecType(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'.

FIXED = <StopSpecType.FIXED: 'FIXED'>
TRAILING = <StopSpecType.TRAILING: 'TRAILING'>
@dataclass
class StopSpec:
StopSpec( type: StopSpecType = <StopSpecType.FIXED: 'FIXED'>, fixed: Optional[FixedStopRule] = None, trailing: Optional[TrailingStopRule] = None)
type: StopSpecType = <StopSpecType.FIXED: 'FIXED'>
fixed: Optional[FixedStopRule] = None
trailing: Optional[TrailingStopRule] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def validateReplacement( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fixedStop( stopLoss: FixedStopRule) -> StopSpec:
@staticmethod
def trailingStop( trailingStop: TrailingStopRule) -> StopSpec:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> StopSpec:
@dataclass
class ProfitLevelCondition:
ProfitLevelCondition( type: ProfitThresholdType = <ProfitThresholdType.PERCENT_GAIN: 'PERCENT_GAIN'>, value: float = 0.0)
value: float = 0.0
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ProfitLevelCondition:
@dataclass
class ProtectiveExitPlan:
ProtectiveExitPlan( executionPolicy: ProtectiveExecutionPolicy = <ProtectiveExecutionPolicy.BEST_AVAILABLE: 'BEST_AVAILABLE'>, initialProtection: Optional[StopSpec] = None, profitTargets: Optional[StagedProfitTargets] = None, protectionAdjustments: List[ProtectionAdjustment] = <factory>, maxHold: Optional[investfly.models.strategy.StrategyDuration] = None, sessionClose: Optional[SessionTimeTrigger] = None)
initialProtection: Optional[StopSpec] = None
profitTargets: Optional[StagedProfitTargets] = None
protectionAdjustments: List[ProtectionAdjustment]
maxHold: Optional[investfly.models.strategy.StrategyDuration] = None
sessionClose: Optional[SessionTimeTrigger] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ProtectiveExitPlan:
@dataclass
class ProtectionAdjustment:
ProtectionAdjustment( id: str, trigger: ProtectionTrigger, replacementStop: StopSpec)
id: str
replacementStop: StopSpec
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None, profitTargets: Optional[StagedProfitTargets] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ProtectionAdjustment:
class ProtectionTriggerType(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'.

TARGET_FILLED = <ProtectionTriggerType.TARGET_FILLED: 'TARGET_FILLED'>
PROFIT_REACHED = <ProtectionTriggerType.PROFIT_REACHED: 'PROFIT_REACHED'>
TIME_ELAPSED = <ProtectionTriggerType.TIME_ELAPSED: 'TIME_ELAPSED'>
@dataclass
class ProtectionTrigger:
ProtectionTrigger( type: ProtectionTriggerType, targetNumber: Optional[int] = None, profitLevel: Optional[ProfitLevelCondition] = None, timeElapsed: Optional[investfly.models.strategy.StrategyDuration] = None)
targetNumber: Optional[int] = None
profitLevel: Optional[ProfitLevelCondition] = None
timeElapsed: Optional[investfly.models.strategy.StrategyDuration] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None, profitTargets: Optional[StagedProfitTargets] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ProtectionTrigger:
class ScheduledWorkflowIntentFamily(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'.

RANKED_PORTFOLIO_ROTATION = <ScheduledWorkflowIntentFamily.RANKED_PORTFOLIO_ROTATION: 'RANKED_PORTFOLIO_ROTATION'>
TARGET_BASKET_REBALANCE = <ScheduledWorkflowIntentFamily.TARGET_BASKET_REBALANCE: 'TARGET_BASKET_REBALANCE'>
RECURRING_SINGLE_ASSET_ENTRY = <ScheduledWorkflowIntentFamily.RECURRING_SINGLE_ASSET_ENTRY: 'RECURRING_SINGLE_ASSET_ENTRY'>
RECURRING_MULTI_ASSET_ENTRY = <ScheduledWorkflowIntentFamily.RECURRING_MULTI_ASSET_ENTRY: 'RECURRING_MULTI_ASSET_ENTRY'>
SCHEDULED_OPTION_ENTRY = <ScheduledWorkflowIntentFamily.SCHEDULED_OPTION_ENTRY: 'SCHEDULED_OPTION_ENTRY'>
def strategySecurityTypeMatches( strategyAssetType: investfly.models.marketdata.SecurityType, selectorSecurityType: investfly.models.marketdata.SecurityType) -> bool:
def isSpotRebalanceAsset(assetType: investfly.models.marketdata.SecurityType) -> bool:
def isConfigStrategyAsset(assetType: investfly.models.marketdata.SecurityType) -> bool:
def isScheduledIntentFamilyApplicableForAsset( family: ScheduledWorkflowIntentFamily, assetType: investfly.models.marketdata.SecurityType) -> bool:
@dataclass
class ScheduledWorkflowIntent:
ScheduledWorkflowIntent( family: Optional[ScheduledWorkflowIntentFamily] = None, securityType: Optional[investfly.models.marketdata.SecurityType] = None, portfolioRotation: Optional[PortfolioRotationConfig] = None, targetBasketRebalance: Optional[TargetBasketRebalanceConfig] = None, recurringMultiAssetEntry: Optional[RecurringMultiAssetEntryConfig] = None, recurringSingleAssetEntry: Optional[RecurringSingleAssetEntryConfig] = None, scheduledOptionEntry: Optional[investfly.models.strategy.OptionScheduledEntryConfig] = None)
family: Optional[ScheduledWorkflowIntentFamily] = None
securityType: Optional[investfly.models.marketdata.SecurityType] = None
portfolioRotation: Optional[PortfolioRotationConfig] = None
targetBasketRebalance: Optional[TargetBasketRebalanceConfig] = None
recurringMultiAssetEntry: Optional[RecurringMultiAssetEntryConfig] = None
recurringSingleAssetEntry: Optional[RecurringSingleAssetEntryConfig] = None
scheduledOptionEntry: Optional[investfly.models.strategy.OptionScheduledEntryConfig] = None
def validate(self) -> None:
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[ScheduledWorkflowIntent]:
@dataclass
class PortfolioRotationConfig:
PortfolioRotationConfig( universe: investfly.models.strategy.SecurityUniverseSelector = <factory>, rankScore: Optional[SecurityScoreExpression] = None, targetCount: int = 10, allocationBudget: Optional[PositionSizeSpec] = None)
rankScore: Optional[SecurityScoreExpression] = None
targetCount: int = 10
allocationBudget: Optional[PositionSizeSpec] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> PortfolioRotationConfig:
@dataclass
class TargetWeightSpec:
TargetWeightSpec(symbol: str, weightPct: float)
symbol: str
weightPct: float
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> TargetWeightSpec:
@dataclass
class TargetBasketRebalanceConfig:
TargetBasketRebalanceConfig( targetWeights: List[TargetWeightSpec] = <factory>, driftTolerancePct: Optional[float] = 5.0, allocationBudget: Optional[PositionSizeSpec] = None)
targetWeights: List[TargetWeightSpec]
driftTolerancePct: Optional[float] = 5.0
allocationBudget: Optional[PositionSizeSpec] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> TargetBasketRebalanceConfig:
class ContributionMode(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'.

FIXED_CASH = <ContributionMode.FIXED_CASH: 'FIXED_CASH'>
BUYING_POWER_PERCENT = <ContributionMode.BUYING_POWER_PERCENT: 'BUYING_POWER_PERCENT'>
CASH_SWEEP = <ContributionMode.CASH_SWEEP: 'CASH_SWEEP'>
@dataclass
class ContributionSpec:
ContributionSpec( mode: ContributionMode = <ContributionMode.FIXED_CASH: 'FIXED_CASH'>, amount: Optional[float] = 100.0, percent: Optional[float] = None, minCashReservePct: Optional[float] = None)
amount: Optional[float] = 100.0
percent: Optional[float] = None
minCashReservePct: Optional[float] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> ContributionSpec:
@dataclass
class RecurringMultiAssetEntryTarget:
RecurringMultiAssetEntryTarget( symbol: Optional[str] = None, side: investfly.models.portfolio.PositionType = LONG, weightPct: Optional[float] = None, execution: Optional[OpenExecutionSettings] = None)
symbol: Optional[str] = None
weightPct: Optional[float] = None
execution: Optional[OpenExecutionSettings] = None
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> RecurringMultiAssetEntryTarget:
@dataclass
class RecurringMultiAssetEntryConfig:
RecurringMultiAssetEntryConfig( targets: List[RecurringMultiAssetEntryTarget] = <factory>, contribution: Optional[ContributionSpec] = None)
contribution: Optional[ContributionSpec] = None
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> RecurringMultiAssetEntryConfig:
@dataclass
class RecurringSingleAssetEntryConfig:
RecurringSingleAssetEntryConfig( symbol: Optional[str] = None, side: investfly.models.portfolio.PositionType = LONG, execution: OpenExecutionSettings = <factory>)
symbol: Optional[str] = None
def validate(self) -> None:
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> RecurringSingleAssetEntryConfig:
@dataclass
class ScheduledJob:
ScheduledJob( id: Optional[str] = None, name: Optional[str] = None, schedule: investfly.models.strategy.TriggerSchedule = <factory>, runWhen: Optional[GuardPolicy] = None, intent: Optional[ScheduledWorkflowIntent] = None)
id: Optional[str] = None
name: Optional[str] = None
runWhen: Optional[GuardPolicy] = None
intent: Optional[ScheduledWorkflowIntent] = None
def validate(self) -> None:
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ScheduledJob:
class MarketRegimeRule(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'.

PRICE_ABOVE_SMA = <MarketRegimeRule.PRICE_ABOVE_SMA: 'PRICE_ABOVE_SMA'>
PRICE_BELOW_SMA = <MarketRegimeRule.PRICE_BELOW_SMA: 'PRICE_BELOW_SMA'>
class MarketRegimeTimeFrame(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'.

DAY = <MarketRegimeTimeFrame.DAY: 'DAY'>
class StrategyGuardType(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'.

MARKET_REGIME = <StrategyGuardType.MARKET_REGIME: 'MARKET_REGIME'>
TIME = <StrategyGuardType.TIME: 'TIME'>
PORTFOLIO_MARGIN = <StrategyGuardType.PORTFOLIO_MARGIN: 'PORTFOLIO_MARGIN'>
CRYPTO_MARKET_REGIME = <StrategyGuardType.CRYPTO_MARKET_REGIME: 'CRYPTO_MARKET_REGIME'>
class TimeWindowMode(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'.

CUSTOM = <TimeWindowMode.CUSTOM: 'CUSTOM'>
RTH_ONLY = <TimeWindowMode.RTH_ONLY: 'RTH_ONLY'>
WEEKDAY_SESSION = <TimeWindowMode.WEEKDAY_SESSION: 'WEEKDAY_SESSION'>
class GuardScope(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'.

SCHEDULED_JOB = <GuardScope.SCHEDULED_JOB: 'SCHEDULED_JOB'>
ENTRY_RULE = <GuardScope.ENTRY_RULE: 'ENTRY_RULE'>
@dataclass(frozen=True)
class StrategyGuardMetadata:
StrategyGuardMetadata( assetTypes: FrozenSet[investfly.models.marketdata.SecurityType], scopes: FrozenSet[GuardScope])
scopes: FrozenSet[GuardScope]
STRATEGY_GUARD_METADATA: Dict[StrategyGuardType, StrategyGuardMetadata] = {<StrategyGuardType.MARKET_REGIME: 'MARKET_REGIME'>: StrategyGuardMetadata(assetTypes=frozenset({OPTION, ETF, STOCK}), scopes=frozenset({<GuardScope.SCHEDULED_JOB: 'SCHEDULED_JOB'>})), <StrategyGuardType.TIME: 'TIME'>: StrategyGuardMetadata(assetTypes=frozenset({OPTION, CRYPTO, FUTURE, ETF, FOREX, STOCK}), scopes=frozenset({<GuardScope.ENTRY_RULE: 'ENTRY_RULE'>})), <StrategyGuardType.PORTFOLIO_MARGIN: 'PORTFOLIO_MARGIN'>: StrategyGuardMetadata(assetTypes=frozenset({OPTION, FOREX, FUTURE}), scopes=frozenset({<GuardScope.ENTRY_RULE: 'ENTRY_RULE'>, <GuardScope.SCHEDULED_JOB: 'SCHEDULED_JOB'>})), <StrategyGuardType.CRYPTO_MARKET_REGIME: 'CRYPTO_MARKET_REGIME'>: StrategyGuardMetadata(assetTypes=frozenset({CRYPTO}), scopes=frozenset({<GuardScope.SCHEDULED_JOB: 'SCHEDULED_JOB'>}))}
def isStrategyGuardTypeApplicableForAssetAndScope( guardType: StrategyGuardType, assetType: investfly.models.marketdata.SecurityType, scope: GuardScope) -> bool:
def strategyGuardTypesForAssetAndScope( assetType: investfly.models.marketdata.SecurityType, scope: GuardScope) -> List[StrategyGuardType]:
class StrategyGuard:
type: ClassVar[StrategyGuardType]
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[StrategyGuard]:
@dataclass
class MarketRegimeGuard(StrategyGuard):
MarketRegimeGuard( referenceSecurity: investfly.models.marketdata.Security = <factory>, rule: MarketRegimeRule = <MarketRegimeRule.PRICE_ABOVE_SMA: 'PRICE_ABOVE_SMA'>, timeFrame: MarketRegimeTimeFrame = <MarketRegimeTimeFrame.DAY: 'DAY'>, lookbackPeriods: int = 200)
type: ClassVar[StrategyGuardType] = <StrategyGuardType.MARKET_REGIME: 'MARKET_REGIME'>
rule: MarketRegimeRule = <MarketRegimeRule.PRICE_ABOVE_SMA: 'PRICE_ABOVE_SMA'>
lookbackPeriods: int = 200
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[MarketRegimeGuard]:
@dataclass
class TimeGuard(StrategyGuard):
TimeGuard( startTime: Optional[str] = None, endTime: Optional[str] = None, days: Optional[investfly.models.strategy.ScheduleDayMode] = None, windowMode: Optional[TimeWindowMode] = None)
type: ClassVar[StrategyGuardType] = <StrategyGuardType.TIME: 'TIME'>
startTime: Optional[str] = None
endTime: Optional[str] = None
windowMode: Optional[TimeWindowMode] = None
@staticmethod
def between( startTime: Optional[str] = '09:30', endTime: Optional[str] = '16:00', days: Optional[investfly.models.strategy.ScheduleDayMode] = None, windowMode: Optional[TimeWindowMode] = None) -> TimeGuard:
def resolvedWindowMode(self) -> TimeWindowMode:
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[TimeGuard]:
@dataclass
class PortfolioMarginGuard(StrategyGuard):
PortfolioMarginGuard( minBuyingPowerPct: Optional[float] = None, minMarginBufferPct: Optional[float] = None, maxLeverage: Optional[float] = None)
type: ClassVar[StrategyGuardType] = <StrategyGuardType.PORTFOLIO_MARGIN: 'PORTFOLIO_MARGIN'>
minBuyingPowerPct: Optional[float] = None
minMarginBufferPct: Optional[float] = None
maxLeverage: Optional[float] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[PortfolioMarginGuard]:
@dataclass
class CryptoMarketRegimeGuard(MarketRegimeGuard):
CryptoMarketRegimeGuard( referenceSecurity: investfly.models.marketdata.Security = <factory>, rule: MarketRegimeRule = <MarketRegimeRule.PRICE_ABOVE_SMA: 'PRICE_ABOVE_SMA'>, timeFrame: MarketRegimeTimeFrame = <MarketRegimeTimeFrame.DAY: 'DAY'>, lookbackPeriods: int = 200)
type: ClassVar[StrategyGuardType] = <StrategyGuardType.CRYPTO_MARKET_REGIME: 'CRYPTO_MARKET_REGIME'>
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[CryptoMarketRegimeGuard]:
Inherited Members
MarketRegimeGuard
referenceSecurity
rule
timeFrame
lookbackPeriods
validate
toDict
@dataclass
class GuardDecision:
GuardDecision( passed: bool, guardType: StrategyGuardType, scope: Optional[GuardScope] = None, reasonCode: Optional[str] = None, message: Optional[str] = None, targetKey: Optional[str] = None)
passed: bool
guardType: StrategyGuardType
scope: Optional[GuardScope] = None
reasonCode: Optional[str] = None
message: Optional[str] = None
targetKey: Optional[str] = None
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> GuardDecision:
@dataclass
class GuardPolicy:
GuardPolicy( guards: List[StrategyGuard] = <factory>)
guards: List[StrategyGuard]
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None, scope: Optional[GuardScope] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[GuardPolicy]:
@dataclass
class OpenPositionSpec:
OpenPositionSpec( side: investfly.models.portfolio.PositionType = LONG, orderSpec: Optional[OrderSpec] = None, positionSize: Optional[PositionSizeSpec] = None, contractSelector: Optional[investfly.models.strategy.FutureContractSelector] = None, allowAddToExisting: Optional[bool] = None)
orderSpec: Optional[OrderSpec] = None
positionSize: Optional[PositionSizeSpec] = None
contractSelector: Optional[investfly.models.strategy.FutureContractSelector] = None
allowAddToExisting: Optional[bool] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> OpenPositionSpec:
@dataclass
class ClosePositionSpec:
ClosePositionSpec( closePercent: Optional[float] = None, orderSpec: Optional[OrderSpec] = None)
closePercent: Optional[float] = None
orderSpec: Optional[OrderSpec] = None
def validate( self, securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> ClosePositionSpec:
class SecuritySelectionSource(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'.

STRATEGY_UNIVERSE = <SecuritySelectionSource.STRATEGY_UNIVERSE: 'STRATEGY_UNIVERSE'>
CUSTOM_LIST = <SecuritySelectionSource.CUSTOM_LIST: 'CUSTOM_LIST'>
@dataclass
class SecuritySelection:
SecuritySelection( source: SecuritySelectionSource = <SecuritySelectionSource.STRATEGY_UNIVERSE: 'STRATEGY_UNIVERSE'>, customSymbols: Optional[Set[str]] = None, candidateFilter: Optional[investfly.models.strategy.SecurityFilterExpression] = None)
customSymbols: Optional[Set[str]] = None
candidateFilter: Optional[investfly.models.strategy.SecurityFilterExpression] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
def collectSecurityExpressions( self, result: List[investfly.models.strategy.SecurityFilterExpression]) -> None:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> SecuritySelection:
@dataclass
class SecurityScoreExpression:
SecurityScoreExpression( dataParams: Optional[Dict[str, investfly.models.strategy.DataParam]] = None, formula: List[str] = <factory>)
dataParams: Optional[Dict[str, investfly.models.strategy.DataParam]] = None
formula: List[str]
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> SecurityScoreExpression:
@dataclass
class RankedSecuritySelection:
RankedSecuritySelection( universe: investfly.models.strategy.SecurityUniverseSelector = <factory>, scoreExpression: SecurityScoreExpression = <factory>, selectTopN: int = 10)
scoreExpression: SecurityScoreExpression
selectTopN: int = 10
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
def collectSecurityExpressions( self, result: List[investfly.models.strategy.SecurityFilterExpression]) -> None:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> RankedSecuritySelection:
class AllocationMode(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'.

PERCENT_OF_PORTFOLIO = <AllocationMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>
FIXED_NOTIONAL = <AllocationMode.FIXED_NOTIONAL: 'FIXED_NOTIONAL'>
FIXED_QUANTITY = <AllocationMode.FIXED_QUANTITY: 'FIXED_QUANTITY'>
@dataclass
class AllocationModel:
AllocationModel( mode: AllocationMode = <AllocationMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>, percentOfPortfolio: Optional[float] = None, fixedNotional: Optional[float] = None, fixedQuantity: Optional[float] = None)
mode: AllocationMode = <AllocationMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>
percentOfPortfolio: Optional[float] = None
fixedNotional: Optional[float] = None
fixedQuantity: Optional[float] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> AllocationModel:
class PositionSizeMode(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'.

PERCENT_OF_PORTFOLIO = <PositionSizeMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>
PERCENT_OF_PORTFOLIO_RISK = <PositionSizeMode.PERCENT_OF_PORTFOLIO_RISK: 'PERCENT_OF_PORTFOLIO_RISK'>
FIXED_RISK_BUDGET = <PositionSizeMode.FIXED_RISK_BUDGET: 'FIXED_RISK_BUDGET'>
FIXED_NOTIONAL = <PositionSizeMode.FIXED_NOTIONAL: 'FIXED_NOTIONAL'>
FIXED_QUANTITY = <PositionSizeMode.FIXED_QUANTITY: 'FIXED_QUANTITY'>
@dataclass
class PositionSizeSpec:
PositionSizeSpec( mode: Optional[PositionSizeMode] = <PositionSizeMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>, value: Optional[float] = None)
mode: Optional[PositionSizeMode] = <PositionSizeMode.PERCENT_OF_PORTFOLIO: 'PERCENT_OF_PORTFOLIO'>
value: Optional[float] = None
def validate(self, label: str = 'PositionSizeSpec') -> None:
def toAllocationModel(self) -> AllocationModel:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> PositionSizeSpec:
@dataclass
class PositionExposureLimit:
PositionExposureLimit( maxNotional: Optional[float] = None, maxPositionPct: Optional[float] = None, maxQuantity: Optional[float] = None)
maxNotional: Optional[float] = None
maxPositionPct: Optional[float] = None
maxQuantity: Optional[float] = None
def validate(self, label: str = 'PositionExposureLimit') -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> PositionExposureLimit:
class MoveUnit(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'.

PERCENT = <MoveUnit.PERCENT: 'PERCENT'>
TICKS = <MoveUnit.TICKS: 'TICKS'>
PIPS = <MoveUnit.PIPS: 'PIPS'>
class MoveBasis(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'.

AVERAGE_ENTRY_PRICE = <MoveBasis.AVERAGE_ENTRY_PRICE: 'AVERAGE_ENTRY_PRICE'>
class ScalingSizeMode(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'.

PERCENT_OF_BASE = <ScalingSizeMode.PERCENT_OF_BASE: 'PERCENT_OF_BASE'>
FIXED_QUANTITY = <ScalingSizeMode.FIXED_QUANTITY: 'FIXED_QUANTITY'>
@dataclass
class MoveCondition:
MoveCondition( unit: Optional[MoveUnit] = <MoveUnit.PERCENT: 'PERCENT'>, threshold: Optional[float] = None, levels: Optional[List[float]] = None, basis: Optional[MoveBasis] = <MoveBasis.AVERAGE_ENTRY_PRICE: 'AVERAGE_ENTRY_PRICE'>)
unit: Optional[MoveUnit] = <MoveUnit.PERCENT: 'PERCENT'>
threshold: Optional[float] = None
levels: Optional[List[float]] = None
basis: Optional[MoveBasis] = <MoveBasis.AVERAGE_ENTRY_PRICE: 'AVERAGE_ENTRY_PRICE'>
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType, label: str) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> MoveCondition:
@dataclass
class ScaleAddPlan:
ScaleAddPlan( maxSteps: Optional[int] = None, minIntervalBetweenAdds: Optional[investfly.models.strategy.StrategyDuration] = None, scalingSizeMode: Optional[ScalingSizeMode] = None, firstStepPctOfBase: Optional[float] = None, quantity: Optional[float] = None, quantities: Optional[List[float]] = None, stepSizeMultiplier: Optional[float] = None)
maxSteps: Optional[int] = None
minIntervalBetweenAdds: Optional[investfly.models.strategy.StrategyDuration] = None
scalingSizeMode: Optional[ScalingSizeMode] = None
firstStepPctOfBase: Optional[float] = None
quantity: Optional[float] = None
quantities: Optional[List[float]] = None
stepSizeMultiplier: Optional[float] = None
def validateForAsset( self, label: str, securityType: Optional[investfly.models.marketdata.SecurityType], allowUnlimitedSteps: bool) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ScaleAddPlan:
@dataclass
class RepeatSignalAddRule:
RepeatSignalAddRule( onlyIfWinning: Optional[bool] = False, addPlan: Optional[ScaleAddPlan] = None)
onlyIfWinning: Optional[bool] = False
addPlan: Optional[ScaleAddPlan] = None
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType]) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> RepeatSignalAddRule:
@dataclass
class WinnerScaleRule:
WinnerScaleRule( favorableMove: Optional[MoveCondition] = None, addPlan: Optional[ScaleAddPlan] = None)
favorableMove: Optional[MoveCondition] = None
addPlan: Optional[ScaleAddPlan] = None
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> WinnerScaleRule:
@dataclass
class LoserScaleLimits:
LoserScaleLimits( maxLossPctBeforeAdd: Optional[float] = None, maxTotalAddNotional: Optional[float] = None)
maxLossPctBeforeAdd: Optional[float] = None
maxTotalAddNotional: Optional[float] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> LoserScaleLimits:
@dataclass
class LoserScaleRule:
LoserScaleRule( adverseMove: Optional[MoveCondition] = None, addPlan: Optional[ScaleAddPlan] = None, limits: Optional[LoserScaleLimits] = None)
adverseMove: Optional[MoveCondition] = None
addPlan: Optional[ScaleAddPlan] = None
limits: Optional[LoserScaleLimits] = None
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> LoserScaleRule:
@dataclass
class ScalingPlan:
ScalingPlan( addToWinner: Optional[WinnerScaleRule] = None, addToLoser: Optional[LoserScaleRule] = None)
addToWinner: Optional[WinnerScaleRule] = None
addToLoser: Optional[LoserScaleRule] = None
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> ScalingPlan:
@dataclass
class PositionManagementRules:
PositionManagementRules( positionExits: Optional[ExitRules] = None, positionScaling: Optional[ScalingPlan] = None, optionGroupExits: Optional[investfly.models.strategy.OptionStrategyRules.OptionExitRules] = None)
positionExits: Optional[ExitRules] = None
positionScaling: Optional[ScalingPlan] = None
optionGroupExits: Optional[investfly.models.strategy.OptionStrategyRules.OptionExitRules] = None
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType) -> None:
def hasScheduledScaling(self) -> bool:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[PositionManagementRules]:
@dataclass
class OptionGroupExposureLimits:
OptionGroupExposureLimits( maxGroupRiskPct: Optional[float] = None, maxGroupNotional: Optional[float] = None, maxContracts: Optional[int] = None)
maxGroupRiskPct: Optional[float] = None
maxGroupNotional: Optional[float] = None
maxContracts: Optional[int] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> OptionGroupExposureLimits:
@dataclass
class PortfolioLimits:
PortfolioLimits( maxOpenPositions: Optional[int] = None, maxPositionExposure: Optional[PositionExposureLimit] = None, maxTotalMarginPct: Optional[float] = None, maxOpenPositionsPerSymbol: Optional[int] = None, optionGroupExposure: Optional[OptionGroupExposureLimits] = None)
maxOpenPositions: Optional[int] = None
maxPositionExposure: Optional[PositionExposureLimit] = None
maxTotalMarginPct: Optional[float] = None
maxOpenPositionsPerSymbol: Optional[int] = None
optionGroupExposure: Optional[OptionGroupExposureLimits] = None
def validate(self) -> None:
def validateForAsset( self, securityType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> PortfolioLimits:
class OrderDuration(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'.

GTC = <OrderDuration.GTC: 'GTC'>
DAY = <OrderDuration.DAY: 'DAY'>
class StrategyOrderType(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'.

MARKET_ORDER = <StrategyOrderType.MARKET_ORDER: 'MARKET_ORDER'>
LIMIT_ORDER = <StrategyOrderType.LIMIT_ORDER: 'LIMIT_ORDER'>
class LimitPriceOffsetUnit(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'.

PERCENT = <LimitPriceOffsetUnit.PERCENT: 'PERCENT'>
PRICE = <LimitPriceOffsetUnit.PRICE: 'PRICE'>
TICKS = <LimitPriceOffsetUnit.TICKS: 'TICKS'>
POINTS = <LimitPriceOffsetUnit.POINTS: 'POINTS'>
PIPS = <LimitPriceOffsetUnit.PIPS: 'PIPS'>
@dataclass
class LimitPriceOffset:
LimitPriceOffset( unit: Optional[LimitPriceOffsetUnit] = None, value: Optional[float] = None)
unit: Optional[LimitPriceOffsetUnit] = None
value: Optional[float] = None
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType], label: str = 'LimitPriceOffset') -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> LimitPriceOffset:
@dataclass
class OrderSpec:
OrderSpec( orderType: Optional[StrategyOrderType] = <StrategyOrderType.MARKET_ORDER: 'MARKET_ORDER'>, orderDuration: Optional[OrderDuration] = <OrderDuration.DAY: 'DAY'>, limitPriceOffset: Optional[LimitPriceOffset] = None, unfilledTimeout: Optional[investfly.models.strategy.StrategyDuration] = None)
orderType: Optional[StrategyOrderType] = <StrategyOrderType.MARKET_ORDER: 'MARKET_ORDER'>
orderDuration: Optional[OrderDuration] = <OrderDuration.DAY: 'DAY'>
limitPriceOffset: Optional[LimitPriceOffset] = None
unfilledTimeout: Optional[investfly.models.strategy.StrategyDuration] = None
def validate( self, label: str = 'OrderSpec', securityType: Optional[investfly.models.marketdata.SecurityType] = None) -> None:
def validateForAsset( self, securityType: Optional[investfly.models.marketdata.SecurityType], label: str = 'OrderSpec') -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> OrderSpec:
class FutureRollMode(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'.

NONE = <FutureRollMode.NONE: 'NONE'>
CLOSE_ONLY = <FutureRollMode.CLOSE_ONLY: 'CLOSE_ONLY'>
ROLL_TO_NEXT = <FutureRollMode.ROLL_TO_NEXT: 'ROLL_TO_NEXT'>
@dataclass
class FutureLifecycleRules:
FutureLifecycleRules( rollMode: FutureRollMode = <FutureRollMode.NONE: 'NONE'>, daysBeforeExpiry: Optional[int] = None)
rollMode: FutureRollMode = <FutureRollMode.NONE: 'NONE'>
daysBeforeExpiry: Optional[int] = None
def validate(self) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[FutureLifecycleRules]:
@dataclass
class AssetLifecycleRules:
AssetLifecycleRules( future: Optional[FutureLifecycleRules] = None, option: Optional[investfly.models.strategy.OptionLifecycleRules] = None)
future: Optional[FutureLifecycleRules] = None
def validateForAsset( self, assetType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Optional[Dict[str, Any]]) -> Optional[AssetLifecycleRules]:
@dataclass
class StrategyConfig:
StrategyConfig( assetType: investfly.models.marketdata.SecurityType, assetLifecycle: Optional[AssetLifecycleRules] = None, entries: EntryRules = <factory>, positionManagement: Optional[PositionManagementRules] = <factory>, scheduledJobs: List[ScheduledJob] = <factory>, portfolioLimits: Optional[PortfolioLimits] = <factory>)
assetLifecycle: Optional[AssetLifecycleRules] = None
entries: EntryRules
positionManagement: Optional[PositionManagementRules]
scheduledJobs: List[ScheduledJob]
portfolioLimits: Optional[PortfolioLimits]
def entryUniverses( self) -> List[investfly.models.strategy.SecurityUniverseSelector]:
def firstEntryUniverse( self) -> Optional[investfly.models.strategy.SecurityUniverseSelector]:
def openPositions(self) -> List[OpenPositionSpec]:
def hasEnabledLogic(self) -> bool:
def firstEntryRule(self) -> Optional[EntryRule]:
def firstSignalExit(self) -> Optional[ExitRule]:
def firstOpenExpression( self) -> Optional[investfly.models.strategy.SecurityFilterExpression]:
def firstCloseExpression( self) -> Optional[investfly.models.strategy.SecurityFilterExpression]:
def firstOpenPosition( self) -> Optional[OpenPositionSpec]:
def securityExpressions( self) -> List[investfly.models.strategy.SecurityFilterExpression]:
def marketDataSecurityExpressions( self) -> List[investfly.models.strategy.SecurityFilterExpression]:
def optionOpenExecutions( self) -> List[OpenExecutionSettings]:
def validate(self) -> None:
def validateForTradedSecurityType( self, tradedSecurityType: investfly.models.marketdata.SecurityType) -> None:
def toDict(self) -> Dict[str, Any]:
@staticmethod
def fromDict( jsonDict: Dict[str, Any]) -> StrategyConfig: