XF
/
fisk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Implement TP/SL price to percent conversion

This commit is contained in:
Mark Veidemanis 2023-01-05 19:27:59 +00:00
parent a6f9e74ee1
commit d3e2bc8648
Signed by: m
GPG Key ID: 5ACFCEED46C0904F
6 changed files with 389 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
core/exchanges/common.py
View File

@ -1,3 +1,5 @@
from decimal import Decimal as D
from core.lib.elastic import store_msg
@ -18,7 +20,78 @@ def get_balance_hook(user_id, user_name, account_id, account_name, balance):
)
def tp_price_to_percent(tp_price, current_price, current_units, unrealised_pl):
# Is this right?
pl_per_unit = D(unrealised_pl) / D(current_units)
initial_price = D(current_price) - pl_per_unit
# Get the percent change of the TP price from the initial price.
change_percent = ((D(tp_price) - initial_price) / initial_price) * 100
# Doesn't check direction
return abs(round(change_percent, 5))
def sl_price_to_percent(sl_price, current_price, current_units, unrealised_pl):
# Is this right?
pl_per_unit = D(unrealised_pl) / D(current_units)
initial_price = D(current_price) - pl_per_unit
# Get the percent change of the SL price from the initial price.
change_percent = ((D(sl_price) - initial_price) / initial_price) * 100
# Doesn't check direction
return abs(round(change_percent, 5))
def convert_open_trades(open_trades):
"""
Convert a list of open trades into a list of Trade-like objects.
"""
trades = []
for trade in open_trades:
current_price = trade["price"]
current_units = trade["currentUnits"]
unrealised_pl = trade["unrealizedPL"]
cast = {
"id": trade["id"],
"symbol": trade["symbol"],
"amount": current_units,
"side": trade["side"],
"state": trade["state"],
"price": current_price,
"pl": unrealised_pl,
}
if "takeProfitOrder" in trade:
if trade["takeProfitOrder"]:
take_profit = trade["takeProfitOrder"]["price"]
take_profit_percent = tp_price_to_percent(
take_profit, current_price, current_units, unrealised_pl
)
cast["take_profit"] = take_profit
cast["take_profit_percent"] = take_profit_percent
if "stopLossOrder" in trade:
if trade["stopLossOrder"]:
stop_loss = trade["stopLossOrder"]["price"]
stop_loss_percent = sl_price_to_percent(
stop_loss, current_price, current_units, unrealised_pl
)
cast["stop_loss"] = stop_loss
cast["stop_loss_percent"] = stop_loss_percent
if "trailingStopLossOrder" in trade:
if trade["trailingStopLossOrder"]:
trailing_stop_loss = trade["trailingStopLossOrder"]["price"]
trailing_stop_loss_percent = sl_price_to_percent(
trailing_stop_loss, current_price, current_units, unrealised_pl
)
cast["trailing_stop_loss"] = trailing_stop_loss
cast["trailing_stop_loss_percent"] = trailing_stop_loss_percent
trades.append(cast)
return trades

8
core/lib/schemas/oanda_s.py
View File

@ -86,6 +86,13 @@ def parse_value(x):
return 0
def parse_current_units_side(x):
if float(x["currentUnits"]) > 0:
return "long"
elif float(x["currentUnits"]) < 0:
return "short"
def parse_side(x):
prevent_hedging(x)
if float(x["long"]["units"]) > 0:
@ -529,6 +536,7 @@ OpenTradesSchema = {
"stopLossOrder": "stopLossOrder",
"trailingStopLossOrder": "trailingStopLossOrder",
"trailingStopValue": "trailingStopValue",
"side": parse_current_units_side,
}
],
),

0
core/tests/exchanges/__init__.py Normal file
View File

229
core/tests/exchanges/test_common.py Normal file
View File

@ -0,0 +1,229 @@
from django.test import TestCase
from core.exchanges.common import sl_price_to_percent, tp_price_to_percent
class CommonTestCase(TestCase):
# TP
def test_tp_price_to_percent_initial_long(self):
"""
Test that the TP price to percent conversion works for long trades.
"""
tp_price = 1.1 # 10%
current_price = 1.0
current_units = 1
unrealised_pl = 0
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_tp_price_to_percent_initial_short(self):
"""
Test that the TP price to percent conversion works for short trades.
"""
tp_price = 0.9 # 10%
current_price = 1.0
current_units = 1
unrealised_pl = 0
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_tp_price_to_percent_change_long(self):
"""
Test that the TP price to percent conversion works for long trades
when the price has changed.
"""
tp_price = 1.2 # 20%
current_price = 1.1 # + 10%
current_units = 1
unrealised_pl = 0.1 # 10%
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
def test_tp_price_to_percent_change_short(self):
"""
Test that the TP price to percent conversion works for short trades
when the price has changed.
"""
tp_price = 0.8 # 20%
current_price = 1.1 # + 10%
current_units = 1
unrealised_pl = 0.1 # 10%
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
# For multiple units
def test_tp_price_to_percent_initial_long_multi(self):
"""
Test that the TP price to percent conversion works for long trades
with multiple units.
"""
tp_price = 1.1 # 10%
current_price = 1.0
current_units = 10
unrealised_pl = 0
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_tp_price_to_percent_initial_short_multi(self):
"""
Test that the TP price to percent conversion works for short trades
with multiple units.
"""
tp_price = 0.9 # 10%
current_price = 1.0
current_units = 10
unrealised_pl = 0
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_tp_price_to_percent_change_long_multi(self):
"""
Test that the TP price to percent conversion works for long trades
when the price has changed, with multiple units.
"""
tp_price = 1.2 # 20%
current_price = 1.1 # + 10%
current_units = 10
unrealised_pl = 1 # 10%
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
def test_tp_price_to_percent_change_short_multi(self):
"""
Test that the TP price to percent conversion works for short trades
when the price has changed, with multiple units.
"""
tp_price = 0.8 # 20%
current_price = 1.1 # + 10%
current_units = 10
unrealised_pl = 1 # 10%
percent = tp_price_to_percent(
tp_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
# SL
def test_SL_price_to_percent_initial_long(self):
"""
Test that the SL price to percent conversion works for long trades.
"""
sl_price = 1.1 # 10%
current_price = 1.0
current_units = 1
unrealised_pl = 0
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_sl_price_to_percent_initial_short(self):
"""
Test that the SL price to percent conversion works for short trades.
"""
sl_price = 0.9 # 10%
current_price = 1.0
current_units = 1
unrealised_pl = 0
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_sl_price_to_percent_change_long(self):
"""
Test that the SL price to percent conversion works for long trades
when the price has changed.
"""
sl_price = 1.2 # 20%
current_price = 1.1 # + 10%
current_units = 1
unrealised_pl = 0.1 # 10%
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
def test_sl_price_to_percent_change_short(self):
"""
Test that the SL price to percent conversion works for short trades
when the price has changed.
"""
sl_price = 0.8 # 20%
current_price = 1.1 # + 10%
current_units = 1
unrealised_pl = 0.1 # 10%
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
# For multiple units
def test_sl_price_to_percent_initial_long_multi(self):
"""
Test that the SL price to percent conversion works for long trades
with multiple units.
"""
sl_price = 1.1 # 10%
current_price = 1.0
current_units = 10
unrealised_pl = 0
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_sl_price_to_percent_initial_short_multi(self):
"""
Test that the SL price to percent conversion works for short trades
with multiple units.
"""
sl_price = 0.9 # 10%
current_price = 1.0
current_units = 10
unrealised_pl = 0
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 10)
def test_sl_price_to_percent_change_long_multi(self):
"""
Test that the SL price to percent conversion works for long trades
when the price has changed, with multiple units.
"""
sl_price = 1.2 # 20%
current_price = 1.1 # + 10%
current_units = 10
unrealised_pl = 1 # 10%
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)
def test_sl_price_to_percent_change_short_multi(self):
"""
Test that the SL price to percent conversion works for short trades
when the price has changed, with multiple units.
"""
sl_price = 0.8 # 20%
current_price = 1.1 # + 10%
current_units = 10
unrealised_pl = 1 # 10%
percent = sl_price_to_percent(
sl_price, current_price, current_units, unrealised_pl
)
self.assertEqual(percent, 20)

60
core/tests/trading/test_live.py
View File

@ -1,7 +1,9 @@
from django.test import TestCase
from core.exchanges.common import convert_open_trades
from core.models import Trade
from core.tests.helpers import ElasticMock, LiveBase
from core.trading.market import get_precision, get_price, get_sl, get_tp
class LiveTradingTestCase(ElasticMock, LiveBase, TestCase):
@ -25,8 +27,11 @@ class LiveTradingTestCase(ElasticMock, LiveBase, TestCase):
# We need some money to place trades
self.assertTrue(balance > 1000)
def open_trade(self):
posted = self.trade.post()
def open_trade(self, trade=None):
if trade:
posted = trade.post()
else:
posted = self.trade.post()
# Check the opened trade
self.assertEqual(posted["type"], "MARKET_ORDER")
self.assertEqual(posted["symbol"], "EUR_USD")
@ -35,10 +40,14 @@ class LiveTradingTestCase(ElasticMock, LiveBase, TestCase):
return posted
def close_trade(self):
# refresh the trade to get the trade id
self.trade.refresh_from_db()
closed = self.account.client.close_trade(self.trade.order_id)
def close_trade(self, trade=None):
if trade:
trade.refresh_from_db()
closed = self.account.client.close_trade(trade.order_id)
else:
# refresh the trade to get the trade id
self.trade.refresh_from_db()
closed = self.account.client.close_trade(self.trade.order_id)
# Check the feedback from closing the trade
self.assertEqual(closed["type"], "MARKET_ORDER")
@ -83,3 +92,42 @@ class LiveTradingTestCase(ElasticMock, LiveBase, TestCase):
"""
Test converting open trades response to Trade-like format.
"""
eur_usd_price = get_price(self.account, "buy", "EUR_USD")
trade_tp = get_tp("buy", 1, eur_usd_price)
trade_sl = get_sl("buy", 2, eur_usd_price)
# trade_tsl = get_sl("buy", 1, eur_usd_price, return_var=True)
# # TP 1% profit
# trade_tp = eur_usd_price * D(1.01)
# # SL 2% loss
# trade_sl = eur_usd_price * D(0.98)
# # TSL 1% loss
# trade_tsl = eur_usd_price * D(0.99)
trade_precision, display_precision = get_precision(self.account, "EUR_USD")
# Round everything to the display precision
trade_tp = round(trade_tp, display_precision)
trade_sl = round(trade_sl, display_precision)
# trade_tsl = round(trade_tsl, display_precision)
complex_trade = Trade.objects.create(
user=self.user,
account=self.account,
symbol="EUR_USD",
time_in_force="FOK",
type="market",
amount=10,
direction="buy",
take_profit=trade_tp,
stop_loss=trade_sl,
# trailing_stop_loss=trade_tsl,
)
posted = self.open_trade(complex_trade)
print("OPENED", posted)
trades = self.account.client.get_all_open_trades()
print("TRADES", trades)
trades_converted = convert_open_trades(trades["itemlist"])
print("TRADES CONVERTED", trades_converted)
closed = self.close_trade(complex_trade)
print("CLOSED", closed)

42
core/trading/market.py
View File

@ -266,6 +266,28 @@ def get_price_bound(direction, strategy, price, current_price):
return price_bound
def get_precision(account, symbol):
instruments = account.instruments
if not instruments:
log.error("No instruments found")
return
# Extract the information for the symbol
instrument = account.client.extract_instrument(instruments, symbol)
if not instrument:
# sendmsg(user, f"Symbol not found: {symbol}", title="Error")
log.error(f"Symbol not found: {symbol}")
return (None, None)
# Get the required precision
try:
trade_precision = instrument["tradeUnitsPrecision"]
display_precision = instrument["displayPrecision"]
return (trade_precision, display_precision)
except KeyError:
# sendmsg(user, f"Precision not found for {symbol}", title="Error")
log.error(f"Precision not found for {symbol}")
return (None, None)
def execute_strategy(callback, strategy, func):
"""
Execute a strategy.
@ -299,11 +321,6 @@ def execute_strategy(callback, strategy, func):
# Refresh account object
strategy.account = Account.objects.get(id=strategy.account.id)
instruments = strategy.account.instruments
if not instruments:
log.error("No instruments found")
return
# Shorten some hook, strategy and callback vars for convenience
user = strategy.user
account = strategy.account
@ -326,18 +343,9 @@ def execute_strategy(callback, strategy, func):
log.error(f"Symbol not supported by account: {symbol}")
return
# Extract the information for the symbol
instrument = strategy.account.client.extract_instrument(instruments, symbol)
if not instrument:
sendmsg(user, f"Symbol not found: {symbol}", title="Error")
log.error(f"Symbol not found: {symbol}")
return
# Get the required precision
try:
trade_precision = instrument["tradeUnitsPrecision"]
display_precision = instrument["displayPrecision"]
except KeyError:
# Get the precision for the symbol
trade_precision, display_precision = get_precision(account, symbol)
if not trade_precision or not display_precision:
sendmsg(user, f"Precision not found for {symbol}", title="Error")
log.error(f"Precision not found for {symbol}")
return