84 lines
5.2 KiB
Python
84 lines
5.2 KiB
Python
import logging
|
|
from position_logic.base_logic import BasePositionLogic
|
|
|
|
class DefaultFlipLogic(BasePositionLogic):
|
|
"""
|
|
The standard "flip-on-signal" logic used by most simple strategies
|
|
(SMA, MA Cross, and even the per-coin Copy Trader signals).
|
|
|
|
- BUY signal: Closes any short, opens a long.
|
|
- SELL signal: Closes any long, opens a short.
|
|
- FLAT signal: Closes any open position.
|
|
"""
|
|
def handle_signal(self, signal_data: dict, current_strategy_positions: dict) -> dict:
|
|
"""
|
|
Processes a BUY, SELL, or FLAT signal and issues the necessary orders
|
|
to flip or open a position.
|
|
"""
|
|
name = self.strategy_name
|
|
params = signal_data['config']['parameters']
|
|
coin = signal_data['coin']
|
|
desired_signal = signal_data['signal']
|
|
signal_price = signal_data.get('signal_price', 0)
|
|
|
|
size = params.get('size')
|
|
leverage_long = int(params.get('leverage_long', 2))
|
|
leverage_short = int(params.get('leverage_short', 2))
|
|
agent_name = signal_data['config'].get("agent", "default").lower()
|
|
|
|
# --- This logic now correctly targets a specific coin ---
|
|
current_position = current_strategy_positions.get(coin)
|
|
new_position_state = None # Return None to close position
|
|
|
|
if desired_signal == "BUY" or desired_signal == "INIT_BUY":
|
|
new_position_state = {"coin": coin, "side": "long", "size": size}
|
|
|
|
if not current_position:
|
|
logging.warning(f"[{name}]-[{coin}] ACTION: Setting leverage to {leverage_long}x and opening LONG.")
|
|
self.send_order(agent_name, "update_leverage", coin, is_buy=True, size=leverage_long)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=True, size=size)
|
|
self.log_trade(strategy=name, coin=coin, action="OPEN_LONG", price=signal_price, size=size, signal=desired_signal)
|
|
|
|
elif current_position['side'] == 'short':
|
|
logging.warning(f"[{name}]-[{coin}] ACTION: Closing SHORT and opening LONG with {leverage_long}x leverage.")
|
|
self.send_order(agent_name, "update_leverage", coin, is_buy=True, size=leverage_long)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=True, size=current_position['size'], reduce_only=True)
|
|
self.log_trade(strategy=name, coin=coin, action="CLOSE_SHORT", price=signal_price, size=current_position['size'], signal=desired_signal)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=True, size=size)
|
|
self.log_trade(strategy=name, coin=coin, action="OPEN_LONG", price=signal_price, size=size, signal=desired_signal)
|
|
|
|
else: # Already long, do nothing
|
|
logging.info(f"[{name}]-[{coin}] INFO: Already LONG, no action taken.")
|
|
new_position_state = current_position # State is unchanged
|
|
|
|
elif desired_signal == "SELL" or desired_signal == "INIT_SELL":
|
|
new_position_state = {"coin": coin, "side": "short", "size": size}
|
|
|
|
if not current_position:
|
|
logging.warning(f"[{name}]-[{coin}] ACTION: Setting leverage to {leverage_short}x and opening SHORT.")
|
|
self.send_order(agent_name, "update_leverage", coin, is_buy=False, size=leverage_short)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=False, size=size)
|
|
self.log_trade(strategy=name, coin=coin, action="OPEN_SHORT", price=signal_price, size=size, signal=desired_signal)
|
|
|
|
elif current_position['side'] == 'long':
|
|
logging.warning(f"[{name}]-[{coin}] ACTION: Closing LONG and opening SHORT with {leverage_short}x leverage.")
|
|
self.send_order(agent_name, "update_leverage", coin, is_buy=False, size=leverage_short)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=False, size=current_position['size'], reduce_only=True)
|
|
self.log_trade(strategy=name, coin=coin, action="CLOSE_LONG", price=signal_price, size=current_position['size'], signal=desired_signal)
|
|
self.send_order(agent_name, "market_open", coin, is_buy=False, size=size)
|
|
self.log_trade(strategy=name, coin=coin, action="OPEN_SHORT", price=signal_price, size=size, signal=desired_signal)
|
|
|
|
else: # Already short, do nothing
|
|
logging.info(f"[{name}]-[{coin}] INFO: Already SHORT, no action taken.")
|
|
new_position_state = current_position # State is unchanged
|
|
|
|
elif desired_signal == "FLAT":
|
|
if current_position:
|
|
logging.warning(f"[{name}]-[{coin}] ACTION: Close {current_position['side']} position.")
|
|
is_buy = current_position['side'] == 'short' # To close a short, we buy
|
|
self.send_order(agent_name, "market_open", coin, is_buy=is_buy, size=current_position['size'], reduce_only=True)
|
|
self.log_trade(strategy=name, coin=coin, action=f"CLOSE_{current_position['side'].upper()}", price=signal_price, size=current_position['size'], signal=desired_signal)
|
|
# new_position_state is already None, which will remove it
|
|
|
|
return new_position_state
|