Up3x1 Investor 戦略

using System;
using System.Linq;
using System.Collections.Generic;

using Ecng.Common;
using Ecng.Collections;
using Ecng.Serialization;

using StockSharp.Algo.Indicators;
using StockSharp.Algo.Strategies;
using StockSharp.BusinessEntities;
using StockSharp.Messages;

namespace StockSharp.Samples.Strategies;

/// <summary>
/// Range breakout strategy based on the Up3x1 Investor expert advisor.
/// </summary>
public class Up3x1InvestorStrategy : Strategy
{
	private readonly StrategyParam<decimal> _rangeThresholdPips;
	private readonly StrategyParam<decimal> _bodyThresholdPips;
	private readonly StrategyParam<decimal> _stopLossPips;
	private readonly StrategyParam<decimal> _takeProfitPips;
	private readonly StrategyParam<decimal> _trailingStopPips;
	private readonly StrategyParam<decimal> _trailingStepPips;
	private readonly StrategyParam<DataType> _candleType;

	private decimal _prevOpen;
	private decimal _prevClose;
	private decimal _prevHigh;
	private decimal _prevLow;
	private bool _hasPreviousCandle;
	private decimal? _entryPrice;
	private decimal _highestPrice;
	private decimal _lowestPrice;
	private decimal? _trailingStopPrice;

	public decimal RangeThresholdPips { get => _rangeThresholdPips.Value; set => _rangeThresholdPips.Value = value; }
	public decimal BodyThresholdPips { get => _bodyThresholdPips.Value; set => _bodyThresholdPips.Value = value; }
	public decimal StopLossPips { get => _stopLossPips.Value; set => _stopLossPips.Value = value; }
	public decimal TakeProfitPips { get => _takeProfitPips.Value; set => _takeProfitPips.Value = value; }
	public decimal TrailingStopPips { get => _trailingStopPips.Value; set => _trailingStopPips.Value = value; }
	public decimal TrailingStepPips { get => _trailingStepPips.Value; set => _trailingStepPips.Value = value; }
	public DataType CandleType { get => _candleType.Value; set => _candleType.Value = value; }

	public Up3x1InvestorStrategy()
	{
		_rangeThresholdPips = Param(nameof(RangeThresholdPips), 2m)
			.SetDisplay("Range Threshold (pips)", "Minimum previous candle range in pips", "Signals");

		_bodyThresholdPips = Param(nameof(BodyThresholdPips), 1m)
			.SetDisplay("Body Threshold (pips)", "Minimum previous candle body in pips", "Signals");

		_stopLossPips = Param(nameof(StopLossPips), 5m)
			.SetDisplay("Stop Loss (pips)", "Distance of protective stop in pips", "Risk");

		_takeProfitPips = Param(nameof(TakeProfitPips), 5m)
			.SetDisplay("Take Profit (pips)", "Distance of profit target in pips", "Risk");

		_trailingStopPips = Param(nameof(TrailingStopPips), 3m)
			.SetDisplay("Trailing Stop (pips)", "Distance kept behind price when trailing", "Risk");

		_trailingStepPips = Param(nameof(TrailingStepPips), 5m)
			.SetDisplay("Trailing Step (pips)", "Increment required to move trailing stop", "Risk");

		_candleType = Param(nameof(CandleType), TimeSpan.FromHours(4).TimeFrame())
			.SetDisplay("Candle Type", "Primary timeframe for signals", "General");
	}

	/// <inheritdoc />
	public override IEnumerable<(Security sec, DataType dt)> GetWorkingSecurities()
	{
		return [(Security, CandleType)];
	}

	/// <inheritdoc />
	protected override void OnReseted()
	{
		base.OnReseted();

		_prevOpen = 0m;
		_prevClose = 0m;
		_prevHigh = 0m;
		_prevLow = 0m;
		_hasPreviousCandle = false;
		ResetPositionTracking();
	}

	/// <inheritdoc />
	protected override void OnStarted2(DateTime time)
	{
		base.OnStarted2(time);

		// Subscribe to the configured timeframe and process finished candles.
		var subscription = SubscribeCandles(CandleType);
		subscription
			.Bind(ProcessCandle)
			.Start();
	}

	private void ProcessCandle(ICandleMessage candle)
	{
		// Work only with fully formed candles to keep logic aligned with the original EA.
		if (candle.State != CandleStates.Finished)
			return;

		// If position was closed externally, reset tracking.
		if (Position == 0 && _entryPrice != null)
			ResetPositionTracking();

		var pipSize = GetPipSize();
		var stopLossDistance = StopLossPips > 0 ? StopLossPips * pipSize : 0m;
		var takeProfitDistance = TakeProfitPips > 0 ? TakeProfitPips * pipSize : 0m;
		var trailingStopDistance = TrailingStopPips > 0 ? TrailingStopPips * pipSize : 0m;
		var trailingStepDistance = TrailingStepPips > 0 ? TrailingStepPips * pipSize : 0m;

		// Manage existing trades before searching for a new signal.
		if (Position != 0 && _entryPrice != null)
		{
			if (ManageOpenPosition(candle, stopLossDistance, takeProfitDistance, trailingStopDistance, trailingStepDistance))
			{
				_prevOpen = candle.OpenPrice;
				_prevClose = candle.ClosePrice;
				_prevHigh = candle.HighPrice;
				_prevLow = candle.LowPrice;
				_hasPreviousCandle = true;
				return;
			}
		}

		// no indicators bound, skip IsFormedAndOnlineAndAllowTrading

		if (Position != 0)
		{
			_prevOpen = candle.OpenPrice;
			_prevClose = candle.ClosePrice;
			_prevHigh = candle.HighPrice;
			_prevLow = candle.LowPrice;
			_hasPreviousCandle = true;
			return;
		}

		var refOpen = _hasPreviousCandle ? _prevOpen : candle.OpenPrice;
		var refClose = _hasPreviousCandle ? _prevClose : candle.ClosePrice;
		var refHigh = _hasPreviousCandle ? _prevHigh : candle.HighPrice;
		var refLow = _hasPreviousCandle ? _prevLow : candle.LowPrice;

		var range = refHigh - refLow;
		var body = Math.Abs(refClose - refOpen);
		var rangeThreshold = RangeThresholdPips * pipSize;
		var bodyThreshold = BodyThresholdPips * pipSize;

		// Bullish setup: strong bullish candle with large range and body.
		if (range > rangeThreshold && body > bodyThreshold && refClose > refOpen)
		{
			BuyMarket();
			InitializePositionTracking(candle.ClosePrice);
		}
		// Bearish setup: strong bearish candle with large range and body.
		else if (range > rangeThreshold && body > bodyThreshold && refClose < refOpen)
		{
			SellMarket();
			InitializePositionTracking(candle.ClosePrice);
		}

		_prevOpen = candle.OpenPrice;
		_prevClose = candle.ClosePrice;
		_prevHigh = candle.HighPrice;
		_prevLow = candle.LowPrice;
		_hasPreviousCandle = true;
	}

	private bool ManageOpenPosition(ICandleMessage candle, decimal stopLossDistance, decimal takeProfitDistance, decimal trailingStopDistance, decimal trailingStepDistance)
	{
		if (_entryPrice == null)
			return false;

		if (Position > 0)
		{
			// Update the highest price reached by the long position.
			_highestPrice = Math.Max(_highestPrice, candle.HighPrice);

			// Check stop loss.
			if (stopLossDistance > 0m && candle.LowPrice <= _entryPrice.Value - stopLossDistance)
			{
				SellMarket();
				ResetPositionTracking();
				return true;
			}

			// Check take profit.
			if (takeProfitDistance > 0m && candle.HighPrice >= _entryPrice.Value + takeProfitDistance)
			{
				SellMarket();
				ResetPositionTracking();
				return true;
			}

			// Update trailing stop level when the move is large enough.
			if (trailingStopDistance > 0m && _highestPrice - _entryPrice.Value >= trailingStopDistance + trailingStepDistance)
			{
				var candidate = _highestPrice - trailingStopDistance;
				if (_trailingStopPrice == null || candidate - _trailingStopPrice.Value >= trailingStepDistance)
				_trailingStopPrice = candidate;
			}

			// Exit if price returned to the trailing stop.
			if (_trailingStopPrice != null && candle.LowPrice <= _trailingStopPrice.Value)
			{
				SellMarket();
				ResetPositionTracking();
				return true;
			}
		}
		else if (Position < 0)
		{
			// Update the lowest price reached by the short position.
			_lowestPrice = Math.Min(_lowestPrice, candle.LowPrice);

			// Check stop loss for short trades.
			if (stopLossDistance > 0m && candle.HighPrice >= _entryPrice.Value + stopLossDistance)
			{
				BuyMarket();
				ResetPositionTracking();
				return true;
			}

			// Check take profit for short trades.
			if (takeProfitDistance > 0m && candle.LowPrice <= _entryPrice.Value - takeProfitDistance)
			{
				BuyMarket();
				ResetPositionTracking();
				return true;
			}

			// Update trailing stop for the short side.
			if (trailingStopDistance > 0m && _entryPrice.Value - _lowestPrice >= trailingStopDistance + trailingStepDistance)
			{
				var candidate = _lowestPrice + trailingStopDistance;
				if (_trailingStopPrice == null || _trailingStopPrice.Value - candidate >= trailingStepDistance)
				_trailingStopPrice = candidate;
			}

			// Exit once the trailing stop is touched.
			if (_trailingStopPrice != null && candle.HighPrice >= _trailingStopPrice.Value)
			{
				BuyMarket();
				ResetPositionTracking();
				return true;
			}
		}

		return false;
	}

	private void InitializePositionTracking(decimal entryPrice)
	{
		// Store entry information to evaluate stops and trailing logic.
		_entryPrice = entryPrice;
		_highestPrice = entryPrice;
		_lowestPrice = entryPrice;
		_trailingStopPrice = null;
	}

	private void ResetPositionTracking()
	{
		_entryPrice = null;
		_highestPrice = 0m;
		_lowestPrice = 0m;
		_trailingStopPrice = null;
	}

	private decimal GetPipSize()
	{
		var step = Security?.PriceStep;
		if (step == null || step == 0m)
			return 1m;

		return step.Value;
	}
}

import clr

clr.AddReference("StockSharp.Messages")
clr.AddReference("StockSharp.Algo")
clr.AddReference("StockSharp.Algo.Indicators")
clr.AddReference("StockSharp.Algo.Strategies")

from System import TimeSpan
from StockSharp.Messages import DataType, CandleStates
from StockSharp.Algo.Strategies import Strategy


class up3x1_investor_strategy(Strategy):
    """Range breakout strategy with SL/TP and trailing stop, based on Up3x1 Investor EA."""

    def __init__(self):
        super(up3x1_investor_strategy, self).__init__()

        self._range_threshold_pips = self.Param("RangeThresholdPips", 2.0) \
            .SetDisplay("Range Threshold (pips)", "Minimum previous candle range in pips", "Signals")
        self._body_threshold_pips = self.Param("BodyThresholdPips", 1.0) \
            .SetDisplay("Body Threshold (pips)", "Minimum previous candle body in pips", "Signals")
        self._stop_loss_pips = self.Param("StopLossPips", 5.0) \
            .SetDisplay("Stop Loss (pips)", "Distance of protective stop in pips", "Risk")
        self._take_profit_pips = self.Param("TakeProfitPips", 5.0) \
            .SetDisplay("Take Profit (pips)", "Distance of profit target in pips", "Risk")
        self._trailing_stop_pips = self.Param("TrailingStopPips", 3.0) \
            .SetDisplay("Trailing Stop (pips)", "Distance kept behind price when trailing", "Risk")
        self._trailing_step_pips = self.Param("TrailingStepPips", 5.0) \
            .SetDisplay("Trailing Step (pips)", "Increment required to move trailing stop", "Risk")
        self._candle_type = self.Param("CandleType", DataType.TimeFrame(TimeSpan.FromHours(4))) \
            .SetDisplay("Candle Type", "Primary timeframe for signals", "General")

        self._prev_open = 0.0
        self._prev_close = 0.0
        self._prev_high = 0.0
        self._prev_low = 0.0
        self._has_prev = False
        self._entry_price = None
        self._highest = 0.0
        self._lowest = 0.0
        self._trailing_stop = None

    @property
    def RangeThresholdPips(self):
        return float(self._range_threshold_pips.Value)
    @property
    def BodyThresholdPips(self):
        return float(self._body_threshold_pips.Value)
    @property
    def StopLossPips(self):
        return float(self._stop_loss_pips.Value)
    @property
    def TakeProfitPips(self):
        return float(self._take_profit_pips.Value)
    @property
    def TrailingStopPips(self):
        return float(self._trailing_stop_pips.Value)
    @property
    def TrailingStepPips(self):
        return float(self._trailing_step_pips.Value)
    @property
    def CandleType(self):
        return self._candle_type.Value

    def _get_pip(self):
        sec = self.Security
        if sec is None or sec.PriceStep is None or float(sec.PriceStep) <= 0:
            return 1.0
        return float(sec.PriceStep)

    def OnStarted2(self, time):
        super(up3x1_investor_strategy, self).OnStarted2(time)
        self._prev_open = 0.0
        self._prev_close = 0.0
        self._prev_high = 0.0
        self._prev_low = 0.0
        self._has_prev = False
        self._reset_tracking()

        subscription = self.SubscribeCandles(self.CandleType)
        subscription.Bind(self.process_candle).Start()

    def process_candle(self, candle):
        if candle.State != CandleStates.Finished:
            return

        if self.Position == 0 and self._entry_price is not None:
            self._reset_tracking()

        pip = self._get_pip()
        sl_dist = self.StopLossPips * pip if self.StopLossPips > 0 else 0.0
        tp_dist = self.TakeProfitPips * pip if self.TakeProfitPips > 0 else 0.0
        trail_dist = self.TrailingStopPips * pip if self.TrailingStopPips > 0 else 0.0
        trail_step = self.TrailingStepPips * pip if self.TrailingStepPips > 0 else 0.0

        h = float(candle.HighPrice)
        lo = float(candle.LowPrice)
        o = float(candle.OpenPrice)
        c = float(candle.ClosePrice)

        # Manage existing position
        if self.Position != 0 and self._entry_price is not None:
            if self._manage_position(candle, sl_dist, tp_dist, trail_dist, trail_step):
                self._prev_open = o
                self._prev_close = c
                self._prev_high = h
                self._prev_low = lo
                self._has_prev = True
                return

        if self.Position != 0:
            self._prev_open = o
            self._prev_close = c
            self._prev_high = h
            self._prev_low = lo
            self._has_prev = True
            return

        ref_o = self._prev_open if self._has_prev else o
        ref_c = self._prev_close if self._has_prev else c
        ref_h = self._prev_high if self._has_prev else h
        ref_lo = self._prev_low if self._has_prev else lo

        rng = ref_h - ref_lo
        body = abs(ref_c - ref_o)
        range_thresh = self.RangeThresholdPips * pip
        body_thresh = self.BodyThresholdPips * pip

        if rng > range_thresh and body > body_thresh and ref_c > ref_o:
            self.BuyMarket()
            self._init_tracking(c)
        elif rng > range_thresh and body > body_thresh and ref_c < ref_o:
            self.SellMarket()
            self._init_tracking(c)

        self._prev_open = o
        self._prev_close = c
        self._prev_high = h
        self._prev_low = lo
        self._has_prev = True

    def _manage_position(self, candle, sl_dist, tp_dist, trail_dist, trail_step):
        if self._entry_price is None:
            return False

        h = float(candle.HighPrice)
        lo = float(candle.LowPrice)

        if self.Position > 0:
            self._highest = max(self._highest, h)

            if sl_dist > 0 and lo <= self._entry_price - sl_dist:
                self.SellMarket()
                self._reset_tracking()
                return True
            if tp_dist > 0 and h >= self._entry_price + tp_dist:
                self.SellMarket()
                self._reset_tracking()
                return True
            if trail_dist > 0 and self._highest - self._entry_price >= trail_dist + trail_step:
                candidate = self._highest - trail_dist
                if self._trailing_stop is None or candidate - self._trailing_stop >= trail_step:
                    self._trailing_stop = candidate
            if self._trailing_stop is not None and lo <= self._trailing_stop:
                self.SellMarket()
                self._reset_tracking()
                return True

        elif self.Position < 0:
            self._lowest = min(self._lowest, lo)

            if sl_dist > 0 and h >= self._entry_price + sl_dist:
                self.BuyMarket()
                self._reset_tracking()
                return True
            if tp_dist > 0 and lo <= self._entry_price - tp_dist:
                self.BuyMarket()
                self._reset_tracking()
                return True
            if trail_dist > 0 and self._entry_price - self._lowest >= trail_dist + trail_step:
                candidate = self._lowest + trail_dist
                if self._trailing_stop is None or self._trailing_stop - candidate >= trail_step:
                    self._trailing_stop = candidate
            if self._trailing_stop is not None and h >= self._trailing_stop:
                self.BuyMarket()
                self._reset_tracking()
                return True

        return False

    def _init_tracking(self, entry):
        self._entry_price = entry
        self._highest = entry
        self._lowest = entry
        self._trailing_stop = None

    def _reset_tracking(self):
        self._entry_price = None
        self._highest = 0.0
        self._lowest = 0.0
        self._trailing_stop = None

    def OnReseted(self):
        super(up3x1_investor_strategy, self).OnReseted()
        self._prev_open = 0.0
        self._prev_close = 0.0
        self._prev_high = 0.0
        self._prev_low = 0.0
        self._has_prev = False
        self._reset_tracking()

    def CreateClone(self):
        return up3x1_investor_strategy()