Modified the Position handling to use long/short instead of buy/sell for side/position_type. Also modified the unit tests for both Portfolio and Position to reflect these changes. Added a basic historical backtesting capability via backtest.py and using CSV tick data for currency pairs.

This commit is contained in:
Michael Halls-Moore
2015-04-17 12:34:31 +01:00
parent d9a7444fc2
commit e74777802b
12 changed files with 519 additions and 192 deletions
+58 -55
View File
@@ -24,116 +24,119 @@ class Portfolio(object):
return self.equity * self.risk_per_trade
def add_new_position(
self, side, market, units, exposure,
add_price, remove_price
self, position_type, market, units,
exposure, bid, ask
):
ps = Position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
self.positions[market] = ps
def add_position_units(
self, market, units, exposure,
add_price, remove_price
self, market, units,
exposure, bid, ask
):
if market not in self.positions:
return False
else:
ps = self.positions[market]
if ps.position_type == "long":
add_price = ask
else:
add_price = bid
new_total_units = ps.units + units
new_total_cost = ps.avg_price*ps.units + add_price*units
ps.exposure += exposure
ps.avg_price = new_total_cost/new_total_units
ps.units = new_total_units
ps.update_position_price(remove_price, exposure)
ps.update_position_price(bid, ask, exposure)
return True
def remove_position_units(
self, market, units, remove_price
self, market, units, bid, ask
):
if market not in self.positions:
return False
else:
ps = self.positions[market]
if ps.position_type == "long":
remove_price = bid
else:
remove_price = ask
ps.units -= units
exposure = Decimal(str(units))
ps.exposure -= exposure
ps.update_position_price(remove_price, exposure)
ps.update_position_price(bid, ask, exposure)
pnl = ps.calculate_pips() * exposure / remove_price
self.balance += pnl.quantize(Decimal("0.01", ROUND_HALF_DOWN))
return True
def close_position(
self, market, remove_price
self, market, bid, ask
):
if market not in self.positions:
return False
else:
ps = self.positions[market]
ps.update_position_price(remove_price, ps.exposure)
ps.update_position_price(bid, ask, ps.exposure)
if ps.position_type == "long":
remove_price = bid
else:
remove_price = ask
pnl = ps.calculate_pips() * ps.exposure / remove_price
self.balance += pnl.quantize(Decimal("0.01", ROUND_HALF_DOWN))
del[self.positions[market]]
return True
def execute_signal(self, signal_event):
def execute_signal(self, signal_event):
side = signal_event.side
market = signal_event.instrument
units = int(self.trade_units)
# Check side for correct bid/ask prices
if side == "buy":
add_price = Decimal(str(self.ticker.cur_ask))
remove_price = Decimal(str(self.ticker.cur_bid))
else:
add_price = Decimal(str(self.ticker.cur_bid))
remove_price = Decimal(str(self.ticker.cur_ask))
exposure = Decimal(str(units))
bid = Decimal(str(self.ticker.cur_bid))
ask = Decimal(str(self.ticker.cur_ask))
# If there is no position, create one
if market not in self.positions:
if side == "buy":
position_type = "long"
else:
position_type = "short"
self.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
order = OrderEvent(market, units, "market", side)
self.events.put(order)
# If a position exists add or remove units
else:
ps = self.positions[market]
# Check if the sides equal
if side == ps.side:
# Add to the position
add_position_units(
market, units, exposure,
add_price, remove_price
)
else:
# Check if the units close out the position
if side == "buy" and ps.position_type == "long":
add_position_units(market, units, exposure, bid, ask)
elif side == "sell" and ps.position_type == "long":
if units == ps.units:
# Close the position
self.close_position(market, remove_price)
order = OrderEvent(market, units, "market", side)
self.events.put(order)
self.close_position(market, bid, ask)
# TODO: Allow units to be added/removed
elif units < ps.units:
# Remove from the position
self.remove_position_units(
market, units, remove_price
)
else: # units > ps.units
# Close the position and add a new one with
# additional units of opposite side
new_units = units - ps.units
self.close_position(market, remove_price)
return
elif units > ps.units:
return
elif side == "buy" and ps.position_type == "short":
if units == ps.units:
self.close_position(market, bid, ask)
# TODO: Allow units to be added/removed
elif units < ps.units:
return
elif units > ps.units:
return
if side == "buy":
new_side = "sell"
else:
new_side = "buy"
new_exposure = Decimal(str(units))
self.add_new_position(
new_side, market, new_units,
new_exposure, add_price, remove_price
)
elif side == "sell" and ps.position_type == "short":
add_position_units(market, units, exposure, bid, ask)
order = OrderEvent(market, units, "market", side)
self.events.put(order)
print "Balance: %0.2f" % self.balance
+166 -91
View File
@@ -18,200 +18,199 @@ class TestPortfolio(unittest.TestCase):
)
def test_add_position_long(self):
side = "LONG"
position_type = "long"
market = "GBP/USD"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51819")
remove_price = Decimal("1.51770")
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
self.assertEquals(ps.side, side)
self.assertEquals(ps.position_type, position_type)
self.assertEquals(ps.market, market)
self.assertEquals(ps.units, units)
self.assertEquals(ps.exposure, exposure)
self.assertEquals(ps.avg_price, add_price)
self.assertEquals(ps.cur_price, remove_price)
self.assertEquals(ps.avg_price, ask)
self.assertEquals(ps.cur_price, bid)
def test_add_position_short(self):
side = "SHORT"
position_type = "short"
market = "GBP/USD"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51770")
remove_price = Decimal("1.51819")
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
self.assertEquals(ps.side, side)
self.assertEquals(ps.position_type, position_type)
self.assertEquals(ps.market, market)
self.assertEquals(ps.units, units)
self.assertEquals(ps.exposure, exposure)
self.assertEquals(ps.avg_price, add_price)
self.assertEquals(ps.cur_price, remove_price)
self.assertEquals(ps.avg_price, bid)
self.assertEquals(ps.cur_price, ask)
def test_add_position_units_long(self):
side = "LONG"
position_type = "long"
market = "GBP/USD"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51819")
remove_price = Decimal("1.51770")
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
apu = self.port.add_position_units(
market, units, exposure,
add_price, remove_price
bid, ask
)
self.assertFalse(apu)
# Add a position and test for real position
market = "GBP/USD"
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
# Test for addition of units
add_price = Decimal("1.51928")
remove_price = Decimal("1.51878")
bid = Decimal("1.51878")
ask = Decimal("1.51928")
apu = self.port.add_position_units(
market, units, exposure,
add_price, remove_price
bid, ask
)
self.assertTrue(apu)
self.assertEqual(ps.avg_price, Decimal("1.518735"))
def test_add_position_units_short(self):
side = "SHORT"
position_type = "short"
market = "GBP/USD"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51770")
remove_price = Decimal("1.51819")
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
apu = self.port.add_position_units(
market, units, exposure,
add_price, remove_price
bid, ask
)
self.assertFalse(apu)
# Add a position and test for real position
market = "GBP/USD"
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
# Test for addition of units
add_price = Decimal("1.51878")
remove_price = Decimal("1.51928")
bid = Decimal("1.51878")
ask = Decimal("1.51928")
apu = self.port.add_position_units(
market, units, exposure,
add_price, remove_price
bid, ask
)
self.assertTrue(apu)
self.assertEqual(ps.avg_price, Decimal("1.51824"))
def test_remove_position_units_long(self):
side = "LONG"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51819")
remove_price = Decimal("1.51770")
position_type = "long"
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
apu = self.port.remove_position_units(
market, units, remove_price
market, units, bid, ask
)
self.assertFalse(apu)
# Add a position and then add units to it
market = "GBP/USD"
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
add_price = Decimal("1.51928")
remove_price = Decimal("1.51878")
bid = Decimal("1.51878")
ask = Decimal("1.51928")
add_units = 8000
add_exposure = Decimal(str(add_units))
apu = self.port.add_position_units(
market, add_units, add_exposure,
add_price, remove_price
bid, ask
)
self.assertEqual(ps.units, 10000)
self.assertEqual(ps.exposure, Decimal("10000.00"))
self.assertEqual(ps.avg_price, Decimal("1.519062"))
# Test removal of (some) of the units
add_price = Decimal("1.52134")
remove_price = Decimal("1.52017")
bid = Decimal("1.52017")
ask = Decimal("1.52134")
remove_units = 3000
rpu = self.port.remove_position_units(
market, remove_units, remove_price
market, remove_units, bid, ask
)
self.assertTrue(rpu)
self.assertEqual(ps.units, 7000)
self.assertEqual(ps.exposure, Decimal("7000.00"))
self.assertEqual(ps.profit_base, Decimal("2.19054"))
self.assertEqual(self.port.balance, Decimal("100002.19"))
def test_remove_position_units_short(self):
side = "SHORT"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51770")
remove_price = Decimal("1.51819")
position_type = "short"
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
apu = self.port.remove_position_units(
market, units, remove_price
market, units, bid, ask
)
self.assertFalse(apu)
# Add a position and then add units to it
market = "GBP/USD"
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
add_price = Decimal("1.51878")
remove_price = Decimal("1.51928")
bid = Decimal("1.51878")
ask = Decimal("1.51928")
add_units = 8000
add_exposure = Decimal(str(add_units))
apu = self.port.add_position_units(
market, add_units, add_exposure,
add_price, remove_price
bid, ask
)
self.assertEqual(ps.units, 10000)
self.assertEqual(ps.exposure, Decimal("10000.00"))
self.assertEqual(ps.avg_price, Decimal("1.518564"))
# Test removal of (some) of the units
add_price = Decimal("1.52017")
remove_price = Decimal("1.52134")
bid = Decimal("1.52017")
ask = Decimal("1.52134")
remove_units = 3000
rpu = self.port.remove_position_units(
market, remove_units, remove_price
market, remove_units, bid, ask
)
self.assertTrue(rpu)
self.assertEqual(ps.units, 7000)
@@ -220,16 +219,16 @@ class TestPortfolio(unittest.TestCase):
self.assertEqual(self.port.balance, Decimal("99994.52"))
def test_close_position_long(self):
side = "LONG"
units = 2000
exposure = Decimal(str(units))
add_price = Decimal("1.51819")
remove_price = Decimal("1.51770")
position_type = "long"
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
cp = self.port.close_position(
market, remove_price
market, bid, ask
)
self.assertFalse(cp)
@@ -237,12 +236,12 @@ class TestPortfolio(unittest.TestCase):
# Will lose money on the spread
market = "GBP/USD"
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
cp = self.port.close_position(
market, remove_price
market, bid, ask
)
self.assertTrue(cp)
self.assertRaises(ps) # Key doesn't exist
@@ -252,38 +251,114 @@ class TestPortfolio(unittest.TestCase):
# close the position. Balance should be as expected
# for a multi-leg transaction.
self.port.add_new_position(
side, market, units, exposure,
add_price, remove_price
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
add_price = Decimal("1.51928")
remove_price = Decimal("1.51878")
# Add 8000 units
bid = Decimal("1.51878")
ask = Decimal("1.51928")
add_units = 8000
add_exposure = Decimal(str(add_units))
apu = self.port.add_position_units(
market, add_units, add_exposure,
add_price, remove_price
market, add_units,
add_exposure, bid, ask
)
self.assertEqual(ps.units, 10000)
self.assertEqual(ps.exposure, Decimal("10000.00"))
self.assertEqual(ps.avg_price, Decimal("1.519062"))
add_price = Decimal("1.52134")
remove_price = Decimal("1.52017")
# Remove 3000 units
bid = Decimal("1.52017")
ask = Decimal("1.52134")
remove_units = 3000
rpu = self.port.remove_position_units(
market, remove_units, remove_price
market, remove_units, bid, ask
)
self.assertEqual(ps.units, 7000)
self.assertEqual(ps.exposure, Decimal("7000.00"))
self.assertEqual(ps.profit_base, Decimal("2.19054"))
self.assertEqual(self.port.balance, Decimal("100001.54"))
# Close the position
cp = self.port.close_position(
market, remove_price
market, bid, ask
)
self.assertTrue(cp)
self.assertRaises(ps) # Key doesn't exist
self.assertEqual(self.port.balance, Decimal("100006.65"))
def test_close_position_short(self):
position_type = "short"
units = Decimal("2000")
exposure = Decimal("2000.00")
bid = Decimal("1.51770")
ask = Decimal("1.51819")
# Test for no position
market = "EUR/USD"
cp = self.port.close_position(
market, bid, ask
)
self.assertFalse(cp)
# Add a position and then close it
# Will lose money on the spread
market = "GBP/USD"
self.port.add_new_position(
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
cp = self.port.close_position(
market, bid, ask
)
self.assertTrue(cp)
self.assertRaises(ps) # Key doesn't exist
self.assertEqual(self.port.balance, Decimal("99999.35"))
# Add 2000, add another 8000, remove 3000 and then
# close the position. Balance should be as expected
# for a multi-leg transaction.
self.port.add_new_position(
position_type, market, units,
exposure, bid, ask
)
ps = self.port.positions[market]
# Add 8000 units
bid = Decimal("1.51878")
ask = Decimal("1.51928")
add_units = 8000
add_exposure = Decimal(str(add_units))
apu = self.port.add_position_units(
market, add_units,
add_exposure, bid, ask
)
self.assertEqual(ps.units, 10000)
self.assertEqual(ps.exposure, Decimal("10000.00"))
self.assertEqual(ps.avg_price, Decimal("1.518564"))
# Remove 3000 units
bid = Decimal("1.52017")
ask = Decimal("1.52134")
remove_units = 3000
rpu = self.port.remove_position_units(
market, remove_units, bid, ask
)
self.assertEqual(ps.units, 7000)
self.assertEqual(ps.exposure, Decimal("7000.00"))
self.assertEqual(ps.profit_base, Decimal("-5.48201"))
self.assertEqual(self.port.balance, Decimal("99993.87"))
# Close the position
cp = self.port.close_position(
market, bid, ask
)
self.assertTrue(cp)
self.assertRaises(ps) # Key doesn't exist
self.assertEqual(self.port.balance, Decimal("99981.08"))
if __name__ == "__main__":
unittest.main()
+20 -9
View File
@@ -3,22 +3,31 @@ from decimal import Decimal, getcontext, ROUND_HALF_DOWN
class Position(object):
def __init__(
self, side, market, units,
exposure, avg_price, cur_price
self, position_type, market,
units, exposure, bid, ask
):
self.side = side
self.position_type = position_type # Long or short
self.market = market
self.units = units
self.exposure = Decimal(str(exposure))
self.avg_price = Decimal(str(avg_price))
self.cur_price = Decimal(str(cur_price))
# Long or short
if self.position_type == "long":
self.avg_price = Decimal(str(ask))
self.cur_price = Decimal(str(bid))
else:
self.avg_price = Decimal(str(bid))
self.cur_price = Decimal(str(ask))
self.profit_base = self.calculate_profit_base(self.exposure)
self.profit_perc = self.calculate_profit_perc(self.exposure)
def calculate_pips(self):
getcontext.prec = 6
mult = Decimal("1")
if self.side == "SHORT":
if self.position_type == "long":
mult = Decimal("1")
elif self.position_type == "short":
mult = Decimal("-1")
return (mult * (self.cur_price - self.avg_price)).quantize(
Decimal("0.00001"), ROUND_HALF_DOWN
@@ -35,8 +44,10 @@ class Position(object):
Decimal("0.00001"), ROUND_HALF_DOWN
)
def update_position_price(self, cur_price, exposure):
self.cur_price = cur_price
def update_position_price(self, bid, ask, exposure):
if self.position_type == "long":
self.cur_price = Decimal(str(bid))
else:
self.cur_price = Decimal(str(ask))
self.profit_base = self.calculate_profit_base(exposure)
self.profit_perc = self.calculate_profit_perc(exposure)
+60 -24
View File
@@ -7,55 +7,91 @@ from position import Position
class TestLongGBPUSDPosition(unittest.TestCase):
def setUp(self):
getcontext.prec = 2
side = "LONG"
position_type = "long"
market = "GBP/USD"
units = Decimal(str(2000))
units = Decimal("2000")
exposure = Decimal("2000.00")
avg_price = Decimal("1.51819")
cur_price = Decimal("1.51770")
bid = Decimal("1.50328")
ask = Decimal("1.50349")
self.position = Position(
side, market, units, exposure,
avg_price, cur_price
position_type, market,
units, exposure, bid, ask
)
def test_calculate_pips(self):
def test_calculate_init_pips(self):
pos_pips = self.position.calculate_pips()
self.assertEqual(pos_pips, Decimal("-0.00049"))
self.assertEqual(pos_pips, Decimal("-0.00021"))
def test_calculate_profit_base(self):
def test_calculate_init_profit_base(self):
profit_base = self.position.calculate_profit_base(self.position.exposure)
self.assertEqual(profit_base, Decimal("-0.64571"))
self.assertEqual(profit_base, Decimal("-0.27939"))
def test_calculate_profit_perc(self):
def test_calculate_init_profit_perc(self):
profit_perc = self.position.calculate_profit_perc(self.position.exposure)
self.assertEqual(profit_perc, Decimal("-0.03229"))
self.assertEqual(profit_perc, Decimal("-0.01397"))
def test_calculate_updated_values(self):
"""
Check that after the bid/ask prices move, that the updated
pips, profit and percentage profit calculations are correct.
"""
bid = Decimal("1.50486")
ask = Decimal("1.50586")
self.position.update_position_price(bid, ask, self.position.exposure)
# Check pips
pos_pips = self.position.calculate_pips()
self.assertEqual(pos_pips, Decimal("0.00137"))
# Check profit base
profit_base = self.position.calculate_profit_base(self.position.exposure)
self.assertEqual(profit_base, Decimal("1.82077"))
# Check profit percentage
profit_perc = self.position.calculate_profit_perc(self.position.exposure)
self.assertEqual(profit_perc, Decimal("0.09104"))
class TestShortGBPUSDPosition(unittest.TestCase):
def setUp(self):
getcontext.prec = 2
side = "SHORT"
position_type = "short"
market = "GBP/USD"
units = 2000
units = Decimal("2000")
exposure = Decimal("2000.00")
avg_price = Decimal("1.51819")
cur_price = Decimal("1.51770")
bid = Decimal("1.50328")
ask = Decimal("1.50349")
self.position = Position(
side, market, units, exposure,
avg_price, cur_price
position_type, market,
units, exposure, bid, ask
)
def test_calculate_pips(self):
def test_calculate_init_pips(self):
pos_pips = self.position.calculate_pips()
self.assertEqual(pos_pips, Decimal("0.00049"))
self.assertEqual(pos_pips, Decimal("-0.00021"))
def test_calculate_profit_base(self):
def test_calculate_init_profit_base(self):
profit_base = self.position.calculate_profit_base(self.position.exposure)
self.assertEqual(profit_base, Decimal("0.64571"))
self.assertEqual(profit_base, Decimal("-0.27935"))
def test_calculate_profit_perc(self):
def test_calculate_init_profit_perc(self):
profit_perc = self.position.calculate_profit_perc(self.position.exposure)
self.assertEqual(profit_perc, Decimal("0.03229"))
self.assertEqual(profit_perc, Decimal("-0.01397"))
def test_calculate_updated_values(self):
"""
Check that after the bid/ask prices move, that the updated
pips, profit and percentage profit calculations are correct.
"""
bid = Decimal("1.50486")
ask = Decimal("1.50586")
self.position.update_position_price(bid, ask, self.position.exposure)
# Check pips
pos_pips = self.position.calculate_pips()
self.assertEqual(pos_pips, Decimal("-0.00258"))
# Check profit base
profit_base = self.position.calculate_profit_base(self.position.exposure)
self.assertEqual(profit_base, Decimal("-3.42661"))
# Check profit percentage
profit_perc = self.position.calculate_profit_perc(self.position.exposure)
self.assertEqual(profit_perc, Decimal("-0.17133"))
if __name__ == "__main__":