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

Compare commits

base: XF:b882ba15d0e1ce031bd344f562f94c2d6cdd2aef
Branches Tags
XF:master
..
compare: XF:d6ab0ffd0eba3fc509d46c3104a3e9a252765b11
Branches Tags
XF:master

No commits in common. "b882ba15d0e1ce031bd344f562f94c2d6cdd2aef" and "d6ab0ffd0eba3fc509d46c3104a3e9a252765b11" have entirely different histories.
b882ba15d0 ... d6ab0ffd0e

4 changed files with 22 additions and 382 deletions
Show Stats Expand all files Collapse all files

2
Makefile
View File

@ -11,7 +11,7 @@ log:
docker-compose --env-file=stack.env logs -f docker-compose --env-file=stack.env logs -f
test: test:
docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py test $(MODULES) -v 2" docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py test $(MODULES)"
migrate: migrate:
docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py migrate" docker-compose --env-file=stack.env run --rm app sh -c ". /venv/bin/activate && python manage.py migrate"

150
core/lib/market.py
View File

@ -8,82 +8,7 @@ from core.util import logs
log = logs.get_logger(__name__) log = logs.get_logger(__name__)
def check_existing_position( def crossfilter(account, symbol, direction, func):
func: str,
position: dict,
open_side: str,
open_symbol: str,
open_units: str,
new_side: str,
new_symbol: str,
trade_side_opposite: str,
):
# Check if we already have a position for the symbol
if open_symbol == new_symbol:
# If the live side is the inverse of what we want to do,
# we can't open a position
if open_side == trade_side_opposite:
# If there is a position open, we can't open a new one in the opposite
# direction
if open_units != "0":
# If we have a short on GBP/AUD, we can only place more shorts on
# GBP/AUD.
if func == "entry":
log.debug(
f"Refusing to open new {new_side} position on {new_symbol} due "
f"to {open_side} position on {open_symbol}"
)
return {
"action": "rejected",
"positions": position,
}
elif func == "exit":
log.debug(
(
f"Found {open_units} units of "
f"{open_symbol} on side {trade_side_opposite}"
)
)
# Pass back opposing side so we can close it
return {
"action": "close",
"side": trade_side_opposite,
"positions": position,
}
return False
def check_conflicting_position(
func: str,
position: dict,
open_base: str,
open_quote: str,
open_side: str,
open_symbol: str,
open_units: str,
new_base: str,
new_quote: str,
new_side: str,
new_symbol: str,
trade_side_opposite: str,
):
if open_base == new_quote or open_quote == new_base:
# If we have a long on GBP/AUD, we can only place shorts on XAU/GBP.
if open_side != trade_side_opposite:
if open_units != "0":
# Only do this for entries
if func == "entry":
log.debug(
f"Refusing to open {new_side} position on {new_symbol} due to "
f"{open_side} position on {open_symbol}"
)
return {
"action": "rejected",
"positions": position,
}
def crossfilter(account, new_symbol, new_direction, func):
""" """
Determine if we are betting against ourselves. Determine if we are betting against ourselves.
Checks open positions for the account, rejecting the trade if there is one Checks open positions for the account, rejecting the trade if there is one
@ -95,11 +20,7 @@ def crossfilter(account, new_symbol, new_direction, func):
:return: dict of action and opposing position, or False :return: dict of action and opposing position, or False
""" """
try: try:
# Only get the data we need position_info = account.client.get_position_info(symbol)
if func == "entry":
all_positions = account.client.get_all_positions()
else:
all_positions = [account.client.get_position_info(new_symbol)]
except GenericAPIError as e: except GenericAPIError as e:
if "No position exists for the specified instrument" in str(e): if "No position exists for the specified instrument" in str(e):
log.debug("No position exists for this symbol") log.debug("No position exists for this symbol")
@ -107,61 +28,28 @@ def crossfilter(account, new_symbol, new_direction, func):
else: else:
log.error(f"Error getting position info: {e}") log.error(f"Error getting position info: {e}")
return None return None
if new_direction == "buy": if direction == "buy":
opposing_side = "short" opposing_side = "short"
new_side = "long" elif direction == "sell":
elif new_direction == "sell":
opposing_side = "long" opposing_side = "long"
new_side = "short"
quotes = [] opposing_position_info = position_info[opposing_side]
new_base, new_quote = new_symbol.split("_") if opposing_position_info["units"] != "0":
for position in all_positions: if func == "entry":
# For Forex, get a list of all the quotes. return {"action": "rejected", "positions": opposing_position_info}
# This is to prevent betting against ourselves. elif func == "exit":
# Consider we have a long position open, EUR/USD, and we want to open a log.debug(
# long position on USD/JPY. If the first goes up, the second one will go (
# down just as much. We won't make any money. f"Found {opposing_position_info['units']} units of "
if "_" in position["symbol"]: f"{symbol} on side {opposing_side}"
open_base, open_quote = position["symbol"].split("_")
quotes.append(open_quote)
open_symbol = position["symbol"]
open_side = position["side"]
open_base, open_quote = open_symbol.split("_")
# Check if we already have a position
existing_position_check = check_existing_position(
func=func,
position=position,
open_side=open_side,
open_symbol=open_symbol,
open_units=position["units"],
new_side=new_side,
new_symbol=new_symbol,
trade_side_opposite=opposing_side,
) )
if existing_position_check:
return existing_position_check
# Check if we are betting against ourselves
conflicting_position_check = check_conflicting_position(
func=func,
position=position,
open_base=open_base,
open_quote=open_quote,
open_side=open_side,
open_symbol=open_symbol,
open_units=position["units"],
new_base=new_base,
new_quote=new_quote,
new_side=new_side,
new_symbol=new_symbol,
trade_side_opposite=opposing_side,
) )
if conflicting_position_check: # Pass back opposing side so we can close it
return conflicting_position_check return {
"action": "close",
"side": opposing_side,
"positions": opposing_position_info,
}
return False return False

0
core/tests/lib/__init__.py
View File

248
core/tests/lib/test_market.py
View File

@ -1,248 +0,0 @@
from django.test import TestCase
from core.lib.market import check_conflicting_position, check_existing_position
class MarketTestCase(TestCase):
def test_conflict_position(self):
position = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "long",
"units": "1",
}
trade_cast = {
"symbol": "GBP_EUR",
"base": "GBP",
"quote": "EUR",
"side": "long",
"units": "1",
}
opposite = "short"
check = check_conflicting_position(
func="entry",
position=position,
open_base=position["base"],
open_quote=position["quote"],
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_base=trade_cast["base"],
new_quote=trade_cast["quote"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertEqual(check["action"], "rejected")
def test_conflict_position_quote(self):
position = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "long",
"units": "1",
}
trade_cast = {
"symbol": "USD_JPY",
"base": "USD",
"quote": "JPY",
"side": "long",
"units": "1",
}
opposite = "short"
check = check_conflicting_position(
func="entry",
position=position,
open_base=position["base"],
open_quote=position["quote"],
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_base=trade_cast["base"],
new_quote=trade_cast["quote"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertEqual(check["action"], "rejected")
def test_conflict_position_invert_allowed(self):
position = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "long",
"units": "1",
}
trade_cast = {
"symbol": "GBP_EUR",
"base": "GBP",
"quote": "EUR",
"side": "short",
"units": "1",
}
opposite = "long"
check = check_conflicting_position(
func="entry",
position=position,
open_base=position["base"],
open_quote=position["quote"],
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_base=trade_cast["base"],
new_quote=trade_cast["quote"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertFalse(check)
def test_conflict_position_quote_invert_allowed(self):
position = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "short",
"units": "1",
}
trade_cast = {
"symbol": "USD_JPY",
"base": "USD",
"quote": "JPY",
"side": "long",
"units": "1",
}
opposite = "short"
check = check_conflicting_position(
func="entry",
position=position,
open_base=position["base"],
open_quote=position["quote"],
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_base=trade_cast["base"],
new_quote=trade_cast["quote"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertFalse(check)
def test_conflict_position_identical_allowed(self):
position = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "long",
"units": "1",
}
trade_cast = {
"symbol": "EUR_USD",
"base": "EUR",
"quote": "USD",
"side": "long",
"units": "1",
}
opposite = "short"
check = check_conflicting_position(
func="entry",
position=position,
open_base=position["base"],
open_quote=position["quote"],
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_base=trade_cast["base"],
new_quote=trade_cast["quote"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertFalse(check)
def test_existing_position_fail(self):
"""
Check we cannot enter a short if we already have a long.
"""
position = {"symbol": "EUR_USD", "side": "long", "units": "1"}
trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
opposite = "long"
check = check_existing_position(
func="entry",
position=position,
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertEqual(check["action"], "rejected")
def test_existing_position_flip_fail(self):
"""
Check we cannot enter a long if we already have a short.
"""
position = {"symbol": "EUR_USD", "side": "short", "units": "1"}
trade_cast = {"symbol": "EUR_USD", "side": "long", "units": "1"}
opposite = "short"
check = check_existing_position(
func="entry",
position=position,
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertEqual(check["action"], "rejected")
def test_existing_position_exit(self):
position = {"symbol": "EUR_USD", "side": "long", "units": "1"}
trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
opposite = "long"
check = check_existing_position(
func="exit",
position=position,
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertEqual(check["action"], "close")
self.assertEqual(check["positions"], position)
self.assertEqual(check["side"], opposite)
def test_existing_position_exit_no_units(self):
position = {"symbol": "EUR_USD", "side": "long", "units": "0"}
trade_cast = {"symbol": "EUR_USD", "side": "short", "units": "1"}
opposite = "long"
check = check_existing_position(
func="exit",
position=position,
open_side=position["side"],
open_symbol=position["symbol"],
open_units=position["units"],
new_side=trade_cast["side"],
new_symbol=trade_cast["symbol"],
trade_side_opposite=opposite,
)
self.assertFalse(check)