investfly.models.strategy.TradingStrategy

Initialize strategy with optional configuration.

Args: config: Dictionary of strategy-specific parameters. Can contain any structure (numeric, list, string, nested dictionaries, etc.). Defaults to None if no configuration is provided.

Note: The dataService and context attributes are set by the execution engine before strategy methods are called. They should not be set manually in the constructor.

Set the data service for accessing market data and indicators.

This method is called by the execution engine to inject the data service into the strategy instance. The data service provides access to:

• Indicator computation (SMA, RSI, MACD, etc.)
• Market data (quotes, bars, financials, news)

Args: dataService: The strategy data service instance. This is a singleton service shared across all strategy executions.

Note: This method is typically called automatically by the execution engine. Strategies should not call this method directly.

Set the execution context for the strategy instance.

This method is called by the execution engine to inject the execution context into the strategy instance. The context contains:

• Portfolio: Current portfolio state with positions and balances
• Universe securities: List of securities in the strategy's trading universe

Args: context: The execution context containing portfolio and universe data.

Note: This method is typically called automatically by the execution engine. Strategies should not call this method directly.

Get the portfolio from the execution context.

Returns: The current portfolio state containing: - Open positions - Account balances - Portfolio performance metrics

Example:

portfolio = self.getPortfolio()
portfolio_value = portfolio.balances.currentValue
for position in portfolio.openPositions:
    print(f"{position.security.symbol}: {position.quantity} shares")

Get the universe securities from the execution context.

Returns: List of Security objects representing all securities in the strategy's trading universe. This is the resolved list based on the security universe selector returned by getSecurityUniverseSelector().

Example:

universe = self.getUniverseSecurities()
for security in universe:
    quote = self.dataService.getQuote(security)
    if quote:
        print(f"{security.symbol}: {quote.lastPrice}")

Return the security universe selector for this strategy.

This method must be implemented by all strategy subclasses. It defines which securities the strategy will evaluate and trade.

Returns: A SecurityUniverseSelector object that defines the strategy's trading universe. The selector can specify: - A single security - A standard predefined list (e.g., S&P 500, NASDAQ 100) - A custom list of symbols - A fundamental query-based selection

Example:

# Single stock
return SecurityUniverseSelector.singleStock("AAPL")

# Standard list
return SecurityUniverseSelector.fromStandardList(StandardSymbolsList.SP_500)

# Custom list
return SecurityUniverseSelector.fromSymbols(
    SecurityType.STOCK, ["AAPL", "MSFT", "GOOGL"]
)

# Fundamental query
query = FinancialQuery(
    SecurityType.STOCK,
    FinancialCondition(FinancialField.MARKET_CAP, ComparisonOperator.GREATER_THAN, 1000000000)
)
return SecurityUniverseSelector.fromFinancialQuery(query)

Handle market data updates (bars/ticks).

This optional method is called whenever market data is updated for securities in the strategy's universe. To enable this callback, override this method and decorate it with @data_trigger.

The @data_trigger decorator specifies when this method should be called:

• type (DataType): The type of market data to trigger on. Currently supported: DataType.BARS (bar/candlestick data).
• barInterval (BarInterval, optional): Required for BARS type. Options: ONE_MINUTE, FIVE_MINUTE, FIFTEEN_MINUTE, THIRTY_MINUTE, SIXTY_MINUTE, ONE_DAY.

Args: updated_securities: List of Security objects that received market data updates triggering this callback. Strategies should iterate over this list instead of the full universe to process only securities with new data.

Returns: Optional list of TradeOrder objects to execute. Return None or an empty list if no trades should be placed.

Note: - If your logic requires Quote data (e.g., LastPrice), use @data_trigger(type=DataType.BARS, barInterval=BarInterval.ONE_MINUTE) to trigger on every 1-minute bar update, then access quotes using self.dataService.getQuote(security). - The method is called only for securities that have received updates, not for the entire universe.

Example:

@data_trigger(type=DataType.BARS, barInterval=BarInterval.ONE_DAY)
def on_market_data(self, updated_securities):
    orders = []
    for security in updated_securities:
        # Compute indicators
        sma = self.dataService.computeIndicatorSeries(
            "SMA", security, {"period": 20, "barInterval": BarInterval.ONE_DAY}
        )
        # Generate trade orders based on indicator values
        if sma.last.value > threshold:
            orders.append(TradeOrder(security, TradeType.BUY, OrderType.MARKET_ORDER))
    return orders if orders else None

Handle scheduled time-based events.

This optional method is called at scheduled intervals. To enable this callback, override this method and decorate it with @time_trigger.

The @time_trigger decorator specifies when this method should be called:

• type (MarketTime): The frequency of scheduled execution. Options: DAILY, HOURLY, WEEKLY, MONTHLY.
• time (str, optional): Specific wall-clock time in "HH:MM:SS" format (24-hour). Only applicable for MarketTime.DAILY. If not specified, executes at market open.

Returns: Optional list of TradeOrder objects to execute. Return None or an empty list if no trades should be placed.

Note: - The time parameter is only valid for MarketTime.DAILY. - Time format must be "HH:MM:SS" in 24-hour format (e.g., "14:30:00" for 2:30 PM). - Multiple @time_trigger decorators can be applied to trigger at multiple times.

Example:

# Execute daily at 10:00 AM
@time_trigger(type=MarketTime.DAILY, time="10:00:00")
def on_schedule(self):
    # Rebalance portfolio at market open
    orders = []
    portfolio = self.getPortfolio()
    # ... rebalancing logic ...
    return orders

# Execute hourly
@time_trigger(type=MarketTime.HOURLY)
def on_schedule(self):
    # Check positions every hour
    return None

Define standard exit conditions for closing positions automatically.

This method provides a convenient way to specify common exit criteria that apply to all positions opened by your strategy. These conditions are evaluated automatically by the platform and executed as market orders when triggered.

Standard Exit Conditions

1. Target Profit: Close position when profit reaches target percentage.

   • Specify as positive percentage (0-100 range).
   • Example: 15.0 means close when position is up 15%.

2. Stop Loss: Close position when loss reaches threshold.

   • Specify as negative percentage.
   • Example: -5.0 means close when position is down 5%.

3. Trailing Stop: Close position when price falls by percentage from peak.

   • Specify as negative percentage.
   • Dynamically adjusts as position becomes profitable.
   • Example: -3.0 means close if price drops 3% from highest point reached.

4. Timeout: Close position after specified time duration.

   • Specify using TimeDelta with TimeUnit (MINUTES, HOURS, DAYS).
   • Example: TimeDelta(value=5, unit=TimeUnit.DAYS) closes after 5 days.

Returns: StandardCloseCriteria object with exit conditions, or None if no standard close conditions should be applied.

Note: - All standard close conditions are executed as MARKET_ORDER for immediate execution. - Multiple conditions can be combined; the first one to trigger closes the position. - Return None (default) if you don't want any standard close conditions. - For more complex exit logic, implement custom logic in on_market_data().

Example:

# Simple stop loss and take profit
def getStandardCloseCondition(self):
    return StandardCloseCriteria(
        targetProfit=10.0,      # Take profit at +10%
        stopLoss=-5.0,          # Stop loss at -5%
        trailingStop=None,
        timeOut=None
    )

# Trailing stop with timeout
def getStandardCloseCondition(self):
    return StandardCloseCriteria(
        targetProfit=None,
        stopLoss=None,
        trailingStop=-3.0,      # Trail 3% from peak
        timeOut=TimeDelta(value=7, unit=TimeUnit.DAYS)
    )

# Comprehensive risk management
def getStandardCloseCondition(self):
    return StandardCloseCriteria(
        targetProfit=20.0,      # Take profit at +20%
        stopLoss=-8.0,         # Stop loss at -8%
        trailingStop=-4.0,     # Trail 4% from peak
        timeOut=TimeDelta(value=30, unit=TimeUnit.DAYS)
    )