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

Continue implementing live risk checks

Browse Source
This commit is contained in:
Mark Veidemanis
2023-01-11 19:46:47 +00:00
parent 93be9e6ffe
commit e55f903f42
12 changed files with 527 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
core/exchanges/common.py
View File

@@ -1,6 +1,10 @@
from decimal import Decimal as D
from core.exchanges import GenericAPIError
from core.lib.elastic import store_msg
from core.util import logs
log = logs.get_logger(__name__)
def get_balance_hook(user_id, user_name, account_id, account_name, balance):
@@ -20,144 +24,73 @@ def get_balance_hook(user_id, user_name, account_id, account_name, balance):
)
def tp_price_to_percent(tp_price, side, current_price, current_units, unrealised_pl):
def get_pair(account, base, quote, invert=False):
"""
Determine the percent change of the TP price from the initial price.
Positive values indicate a profit, negative values indicate a loss.
Get the pair for the given account and currencies.
:param account: Account object
:param base: Base currency
:param quote: Quote currency
:param invert: Invert the pair
:return: currency symbol, e.g. BTC_USD, BTC/USD, etc.
"""
pl_per_unit = D(unrealised_pl) / D(current_units)
if side == "long":
initial_price = D(current_price) - pl_per_unit
# Currently we only have two exchanges with different pair separators
if account.exchange == "alpaca":
separator = "/"
elif account.exchange == "oanda":
separator = "_"
# Flip the pair if needed
if invert:
symbol = f"{quote.upper()}{separator}{base.upper()}"
else:
initial_price = D(current_price) + pl_per_unit
# Get the percent change of the TP price from the initial price.
change_percent = ((initial_price - D(tp_price)) / initial_price) * 100
if side == "long":
if D(tp_price) < initial_price:
loss = True
else:
loss = False
else:
if D(tp_price) > initial_price:
loss = True
else:
loss = False
# if we are in loss on the short side, we want to show a negative
if loss:
change_percent = 0 - abs(change_percent)
else:
change_percent = abs(change_percent)
return round(change_percent, 5)
symbol = f"{base.upper()}{separator}{quote.upper()}"
# Check it exists
if symbol not in account.supported_symbols:
return False
return symbol
def sl_price_to_percent(sl_price, side, current_price, current_units, unrealised_pl):
def to_currency(direction, account, amount, from_currency, to_currency):
"""
Determine the percent change of the SL price from the initial price.
Positive values indicate a loss, negative values indicate a profit.
This may seem backwards, but it is important to note that by default,
SL indicates a loss, and positive values should be expected.
Negative values indicate a negative loss, so a profit.
Convert an amount from one currency to another.
:param direction: Direction of the trade
:param account: Account object
:param amount: Amount to convert
:param from_currency: Currency to convert from
:param to_currency: Currency to convert to
:return: Converted amount
"""
pl_per_unit = D(unrealised_pl) / D(current_units)
if side == "long":
initial_price = D(current_price) - pl_per_unit
else:
initial_price = D(current_price) + pl_per_unit
# If we're converting to the same currency, just return the amount
if from_currency == to_currency:
return amount
inverted = False
# initial_price = D(current_price) - pl_per_unit
# This is needed because OANDA has different values for bid and ask
if direction == "buy":
price_index = "bids"
elif direction == "sell":
price_index = "asks"
symbol = get_pair(account, from_currency, to_currency)
if not symbol:
symbol = get_pair(account, from_currency, to_currency, invert=True)
inverted = True
# Get the percent change of the SL price from the initial price.
change_percent = ((initial_price - D(sl_price)) / initial_price) * 100
# Bit of a hack but it works
if not symbol:
log.error(f"Could not find symbol for {from_currency} -> {to_currency}")
raise Exception("Could not find symbol")
try:
prices = account.client.get_currencies([symbol])
except GenericAPIError as e:
log.error(f"Error getting currencies and inverted currencies: {e}")
return None
price = D(prices["prices"][0][price_index][0]["price"])
# If the trade is long, the SL price will be higher than the initial price.
# if side == "long":
# change_percent *= -1
# If we had to flip base and quote, we need to use the reciprocal of the price
if inverted:
price = D(1.0) / price
if side == "long":
if D(sl_price) > initial_price:
profit = True
else:
profit = False
else:
if D(sl_price) < initial_price:
profit = True
else:
profit = False
# Convert the amount to the destination currency
converted = D(amount) * price
if profit:
change_percent = 0 - abs(change_percent)
else:
change_percent = abs(change_percent)
return 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"]
side = trade["side"]
cast = {
"id": trade["id"],
"symbol": trade["symbol"],
"amount": current_units,
"side": side,
"state": trade["state"],
"price": current_price,
"pl": unrealised_pl,
}
# Add some extra fields, sometimes we have already looked up the
# prices and don't need to call convert_trades_to_usd
if "take_profit_usd" in trade:
cast["take_profit_usd"] = trade["take_profit_usd"]
if "stop_loss_usd" in trade:
cast["stop_loss_usd"] = trade["stop_loss_usd"]
if "trailing_stop_loss_usd" in trade:
cast["trailing_stop_loss_usd"] = trade["trailing_stop_loss_usd"]
if "takeProfitOrder" in trade:
if trade["takeProfitOrder"]:
take_profit = trade["takeProfitOrder"]["price"]
take_profit_percent = tp_price_to_percent(
take_profit, side, 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, side, 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,
side,
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
return converted

211
core/exchanges/convert.py Normal file
View File

@@ -0,0 +1,211 @@
from decimal import Decimal as D
from core.models import Account
from core.trading.market import direction_to_side, get_price, side_to_direction
# Separate module to prevent circular import from
# models -> exchanges -> common -> models
# Since we need Account here to look up missing prices
def tp_price_to_percent(tp_price, side, current_price, current_units, unrealised_pl):
"""
Determine the percent change of the TP price from the initial price.
Positive values indicate a profit, negative values indicate a loss.
"""
pl_per_unit = D(unrealised_pl) / D(current_units)
if side == "long":
initial_price = D(current_price) - pl_per_unit
else:
initial_price = D(current_price) + pl_per_unit
# Get the percent change of the TP price from the initial price.
change_percent = ((initial_price - D(tp_price)) / initial_price) * 100
if side == "long":
if D(tp_price) < initial_price:
loss = True
else:
loss = False
else:
if D(tp_price) > initial_price:
loss = True
else:
loss = False
# if we are in loss on the short side, we want to show a negative
if loss:
change_percent = 0 - abs(change_percent)
else:
change_percent = abs(change_percent)
return round(change_percent, 5)
def sl_price_to_percent(sl_price, side, current_price, current_units, unrealised_pl):
"""
Determine the percent change of the SL price from the initial price.
Positive values indicate a loss, negative values indicate a profit.
This may seem backwards, but it is important to note that by default,
SL indicates a loss, and positive values should be expected.
Negative values indicate a negative loss, so a profit.
"""
pl_per_unit = D(unrealised_pl) / D(current_units)
if side == "long":
initial_price = D(current_price) - pl_per_unit
else:
initial_price = D(current_price) + pl_per_unit
# initial_price = D(current_price) - pl_per_unit
# Get the percent change of the SL price from the initial price.
change_percent = ((initial_price - D(sl_price)) / initial_price) * 100
# If the trade is long, the SL price will be higher than the initial price.
# if side == "long":
# change_percent *= -1
if side == "long":
if D(sl_price) > initial_price:
profit = True
else:
profit = False
else:
if D(sl_price) < initial_price:
profit = True
else:
profit = False
if profit:
change_percent = 0 - abs(change_percent)
else:
change_percent = abs(change_percent)
return round(change_percent, 5)
def annotate_trade_tp_sl_percent(trade):
"""
Annotate the trade with the TP and SL percent.
This works on Trade objects, which will require an additional market
lookup to get the current price.
"""
if "current_price" in trade:
current_price = trade["current_price"]
else:
account_id = trade["account_id"]
account = Account.get_by_id_no_user_check(account_id)
current_price = get_price(account, trade["direction"], trade["symbol"])
trade["current_price"] = current_price
current_units = trade["amount"]
if "pl" in trade:
unrealised_pl = trade["pl"]
else:
unrealised_pl = 0
if "side" in trade:
side = trade["side"]
direction = side_to_direction(side)
trade["direction"] = direction
else:
direction = trade["direction"]
side = direction_to_side(direction)
trade["side"] = side
if "take_profit" in trade:
if trade["take_profit"]:
take_profit = trade["take_profit"]
take_profit_percent = tp_price_to_percent(
take_profit, trade["side"], current_price, current_units, unrealised_pl
)
trade["take_profit_percent"] = take_profit_percent
if "stop_loss" in trade:
if trade["stop_loss"]:
stop_loss = trade["stop_loss"]
stop_loss_percent = sl_price_to_percent(
stop_loss, side, current_price, current_units, unrealised_pl
)
trade["stop_loss_percent"] = stop_loss_percent
if "trailing_stop_loss" in trade:
if trade["trailing_stop_loss"]:
trailing_stop_loss = trade["trailing_stop_loss"]
trailing_stop_loss_percent = sl_price_to_percent(
trailing_stop_loss,
trade["side"],
current_price,
current_units,
unrealised_pl,
)
trade["trailing_stop_loss_percent"] = trailing_stop_loss_percent
return trade
def open_trade_to_unified_format(trade):
"""
Convert an open trade to a Trade-like format.
"""
current_price = trade["price"]
current_units = trade["currentUnits"]
unrealised_pl = trade["unrealizedPL"]
side = trade["side"]
cast = {
"id": trade["id"],
"symbol": trade["symbol"],
"amount": current_units,
"side": side,
"direction": side_to_direction(side),
"state": trade["state"],
"current_price": current_price,
"pl": unrealised_pl,
}
# Add some extra fields, sometimes we have already looked up the
# prices and don't need to call convert_trades_to_usd
# This is mostly for tests, but it can be useful in other places.
if "take_profit_usd" in trade:
cast["take_profit_usd"] = trade["take_profit_usd"]
if "stop_loss_usd" in trade:
cast["stop_loss_usd"] = trade["stop_loss_usd"]
if "trailing_stop_loss_usd" in trade:
cast["trailing_stop_loss_usd"] = trade["trailing_stop_loss_usd"]
if "takeProfitOrder" in trade:
if trade["takeProfitOrder"]:
take_profit = trade["takeProfitOrder"]["price"]
cast["take_profit"] = take_profit
if "stopLossOrder" in trade:
if trade["stopLossOrder"]:
stop_loss = trade["stopLossOrder"]["price"]
cast["stop_loss"] = stop_loss
if "trailingStopLossOrder" in trade:
if trade["trailingStopLossOrder"]:
trailing_stop_loss = trade["trailingStopLossOrder"]["price"]
cast["trailing_stop_loss"] = trailing_stop_loss
return cast
def convert_trades(open_trades):
"""
Convert a list of open trades into a list of Trade-like objects.
"""
trades = []
for trade in open_trades:
if "currentUnits" in trade: # Open trade
cast = open_trade_to_unified_format(trade)
cast = annotate_trade_tp_sl_percent(cast)
trades.append(cast)
else:
cast = annotate_trade_tp_sl_percent(trade)
trades.append(cast)
return trades

13
core/exchanges/oanda.py
View File

@@ -36,18 +36,25 @@ class OANDAExchange(BaseExchange):
response = self.get_instruments()
return [x["name"] for x in response["itemlist"]]
def get_balance(self):
def get_balance(self, return_usd=False):
r = accounts.AccountSummary(self.account_id)
response = self.call(r)
print("RESPONSE", response)
balance = float(response["balance"])
currency = response["currency"]
balance_usd = common.to_currency("sell", self.account, balance, currency, "USD")
print("BALANCE", balance)
print("BALANCE USD", balance_usd)
common.get_balance_hook(
self.account.user.id,
self.account.user.username,
self.account.id,
self.account.name,
balance,
balance_usd,
)
if return_usd:
return balance_usd
return balance
def get_market_value(self, symbol):
@@ -138,7 +145,7 @@ class OANDAExchange(BaseExchange):
def get_all_open_trades(self):
r = trades.OpenTrades(accountID=self.account_id)
return self.call(r)
return self.call(r)["itemlist"]
def close_position(self, side, symbol):
data = {