From 5bce8efd847cf4bd62dd711d69d9e311d61ff2c2 Mon Sep 17 00:00:00 2001 From: Andy Geach Date: Thu, 16 Apr 2026 10:00:48 +0100 Subject: [PATCH 1/2] integration test timer --- .github/workflows/integration-test.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 9f7902a..0733a86 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -3,7 +3,7 @@ name: trading-ig integration test on: schedule: - - cron: '25 2 * * 1-4' + - cron: '0 23 * * 1-4' workflow_dispatch: jobs: From 04deffa71f6bfa35a0ac4102047aae3385684287 Mon Sep 17 00:00:00 2001 From: Andy Geach Date: Fri, 14 Aug 2026 12:28:37 +0100 Subject: [PATCH 2/2] pin ruff version. fix (most) ruff checks. delete old lightstreamer module, update docs --- docs/source/quickstart.rst | 16 +- pyproject.toml | 5 +- sample/all_nodes.py | 5 +- sample/rest.ipynb | 6 +- sample/rest_ig.py | 15 +- sample/sample_ticker.py | 5 +- sample/sample_ticker_rich.py | 5 +- sample/sample_utils.py | 2 +- sample/stream_ig.py | 16 +- tests/retry_test.py | 10 +- tests/test_accounts.py | 6 +- tests/test_activities.py | 12 +- tests/test_dealing.py | 6 +- tests/test_historical_prices.py | 14 +- tests/test_historical_prices_flat.py | 10 +- tests/test_integration.py | 73 ++--- tests/test_positions.py | 6 +- tests/test_rate_limiter.py | 6 +- tests/test_session.py | 10 +- trading_ig/__init__.py | 6 - trading_ig/config.py | 18 +- trading_ig/lightstreamer.py | 467 --------------------------- trading_ig/rest.py | 167 +++++----- trading_ig/stream.py | 15 +- trading_ig/streamer/manager.py | 8 +- trading_ig/streamer/objects.py | 7 +- trading_ig/streamer/ticker.py | 34 +- trading_ig/utils.py | 27 +- 28 files changed, 252 insertions(+), 725 deletions(-) delete mode 100644 trading_ig/lightstreamer.py diff --git a/docs/source/quickstart.rst b/docs/source/quickstart.rst index b5f456a..4dccf46 100644 --- a/docs/source/quickstart.rst +++ b/docs/source/quickstart.rst @@ -49,7 +49,7 @@ Connection >>> from trading_ig.rest import IGService >>> from trading_ig.config import config ->>> ig_service = IGService(config.username, config.password, config.api_key) +>>> ig_service = IGService(config.username, config.password, config.api_key, config.acc_number) >>> ig = ig_service.create_session() >>> ig @@ -108,17 +108,19 @@ Assuming config as above >>> from trading_ig import IGService, IGStreamService >>> from trading_ig.config import config ->>> from trading_ig.lightstreamer import Subscription +>>> from lightstreamer.client import Subscription, SubscriptionListener, ItemUpdate ->>> def on_update(item): ->>> print("{UPDATE_TIME:<8} {stock_name:<19} Bid {BID:>5} Ask {OFFER:>5}".format(stock_name=item["name"], **item["values"])) +>>> class PriceListener(SubscriptionListener): +>>> def onItemUpdate(self, update: ItemUpdate): +>>> logger.info(f"{update.getItemName()} Bid: {update.getValue('BIDPRICE1')}, Offer: {update.getValue('ASKPRICE1')}") >>> ig_service = IGService(config.username, config.password, config.api_key, config.acc_type, acc_number=config.acc_number) >>> ig_stream_service = IGStreamService(ig_service) >>> ig_stream_service.create_session() ->>> sub = Subscription(mode="MERGE", items=["L1:CS.D.GBPUSD.TODAY.IP"], fields=["UPDATE_TIME", "BID", "OFFER"]) ->>> sub.addlistener(on_update) ->>> ig_stream_service.ls_client.subscribe(sub) +>>> sub = Subscription(mode="MERGE", items=[f"PRICE:{config.acc_number}:CS.D.GBPUSD.TODAY.IP"], fields=["BIDPRICE1", "ASKPRICE1"]) +>>> sub.setDataAdapter("Pricing") +>>> sub.addListener(PriceListener()) +>>> ig_stream_service.subscribe(sub) >>> ig_stream_service.disconnect() diff --git a/pyproject.toml b/pyproject.toml index f5067db..67c9349 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,14 +59,17 @@ Issues = "https://github.com/ig-python/trading-ig/issues" [dependency-groups] dev = [ - "ruff", "pytest>=8,<9", "responses>=0.25,<0.26", "coveralls>=3,<4", "sphinx", "sphinx-rtd-theme", + "ruff==0.16.1", ] [tool.uv.build-backend] module-name = "trading_ig" module-root = "" + +[tool.ruff.lint] +ignore = ["B017", "B023", "BLE001", "TRY002" ] diff --git a/sample/all_nodes.py b/sample/all_nodes.py index 7c5d015..7e5e0c6 100644 --- a/sample/all_nodes.py +++ b/sample/all_nodes.py @@ -1,6 +1,7 @@ -from trading_ig.rest import IGService, ApiExceededException +from tenacity import Retrying, retry_if_exception_type, wait_exponential + from trading_ig.config import config -from tenacity import Retrying, wait_exponential, retry_if_exception_type +from trading_ig.rest import ApiExceededException, IGService DEFAULT_RETRY = Retrying( wait=wait_exponential(), retry=retry_if_exception_type(ApiExceededException) diff --git a/sample/rest.ipynb b/sample/rest.ipynb index f9f06c2..18fb156 100644 --- a/sample/rest.ipynb +++ b/sample/rest.ipynb @@ -19,8 +19,8 @@ "metadata": {}, "outputs": [], "source": [ - "from trading_ig.rest import IGService\n", "from trading_ig.config import config\n", + "from trading_ig.rest import IGService\n", "\n", "service = IGService(\n", " config.username,\n", @@ -126,9 +126,9 @@ "metadata": {}, "outputs": [], "source": [ - "from datetime import datetime, timedelta\n", + "from datetime import datetime, timedelta, timezone\n", "\n", - "to_date = datetime.now()\n", + "to_date = datetime.now(timezone.utc)\n", "from_date = to_date - timedelta(days=7)\n", "\n", "service.fetch_transaction_history(from_date=from_date, to_date=to_date)" diff --git a/sample/rest_ig.py b/sample/rest_ig.py index af8116b..c9a2b3b 100644 --- a/sample/rest_ig.py +++ b/sample/rest_ig.py @@ -1,19 +1,18 @@ -#!/usr/bin/env python -# -*- coding:utf-8 -*- - """ IG Markets REST API sample with Python 2015 FemtoTrader """ -from trading_ig import IGService -from trading_ig.config import config import logging # if you need to cache to DB your requests from datetime import timedelta + import requests_cache +from trading_ig import IGService +from trading_ig.config import config + logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) @@ -49,7 +48,7 @@ def main(): # ig_stream_service.create_session(version='3') accounts = ig_service.fetch_accounts() - print("accounts:\n%s" % accounts) + print(f"accounts:\n{accounts}") # account_info = ig_service.switch_account(config.acc_number, False) # print(account_info) @@ -57,12 +56,12 @@ def main(): # open_positions = ig_service.fetch_open_positions() # print("open_positions:\n%s" % open_positions) - print("") + print() # working_orders = ig_service.fetch_working_orders() # print("working_orders:\n%s" % working_orders) - print("") + print() # epic = 'CS.D.EURUSD.MINI.IP' epic = "IX.D.ASX.IFM.IP" # US (SPY) - mini diff --git a/sample/sample_ticker.py b/sample/sample_ticker.py index 99bcd6b..bd158d8 100644 --- a/sample/sample_ticker.py +++ b/sample/sample_ticker.py @@ -1,9 +1,10 @@ import logging import time + +from sample.sample_utils import crypto_epics # fx_epics, index_epics, weekend_epics from trading_ig import IGService, IGStreamService from trading_ig.config import config from trading_ig.streamer.manager import StreamingManager -from sample.sample_utils import crypto_epics # fx_epics, index_epics, weekend_epics def main(): @@ -26,7 +27,7 @@ def main(): sm.start_tick_subscription(epic) tickers.append(sm.ticker(epic)) - for idx in range(0, 10): + for idx in range(10): for ticker in tickers: print(ticker) time.sleep(0.5) diff --git a/sample/sample_ticker_rich.py b/sample/sample_ticker_rich.py index 3da289d..e58e1f5 100644 --- a/sample/sample_ticker_rich.py +++ b/sample/sample_ticker_rich.py @@ -1,13 +1,14 @@ import logging import time + +from sample.sample_utils import crypto_epics # fx_epics, index_epics, weekend_epics from trading_ig import IGService, IGStreamService from trading_ig.config import config from trading_ig.streamer.manager import StreamingManager -from sample.sample_utils import crypto_epics # fx_epics, index_epics, weekend_epics try: - from rich.table import Table from rich.live import Live + from rich.table import Table except ImportError: print("Rich must be installed for this sample") diff --git a/sample/sample_utils.py b/sample/sample_utils.py index 42e1f7d..11ba5fd 100644 --- a/sample/sample_utils.py +++ b/sample/sample_utils.py @@ -1,5 +1,5 @@ def wait_for_input(): - input("{0:-^80}\n".format("HIT CR TO UNSUBSCRIBE AND DISCONNECT")) + input("{:-^80}\n".format("HIT CR TO UNSUBSCRIBE AND DISCONNECT")) # sample weekend spreadbet epics diff --git a/sample/stream_ig.py b/sample/stream_ig.py index f3f8919..684f58d 100755 --- a/sample/stream_ig.py +++ b/sample/stream_ig.py @@ -1,17 +1,19 @@ -import sys +#!/usr/bin/env python + import logging -from datetime import datetime +import sys +from datetime import datetime, timezone from lightstreamer.client import ( + ClientListener, + ItemUpdate, Subscription, SubscriptionListener, - ItemUpdate, - ClientListener, ) +from sample.sample_utils import crypto_epics, wait_for_input from trading_ig import IGService, IGStreamService from trading_ig.config import config -from sample.sample_utils import crypto_epics, wait_for_input logger = logging.getLogger(__name__) @@ -100,7 +102,7 @@ def ig_stream_sample(): class PriceListener(SubscriptionListener): def onItemUpdate(self, update: ItemUpdate): logger.info( - f"{datetime.fromtimestamp(int(update.getValue('TIMESTAMP')) / 1000).strftime('%Y-%m-%d %H:%M:%S')} " + f"{datetime.fromtimestamp(int(update.getValue('TIMESTAMP')) / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} " f"{update.getItemName()} " f"Bid: {update.getValue('BIDPRICE1')}, " f"Offer: {update.getValue('ASKPRICE1')}, " @@ -164,7 +166,7 @@ def onUnsubscription(self): class StatusListener(ClientListener): def onStatusChange(self, status): - print(f"{datetime.now()}: ***** {status} *****") + print(f"{datetime.now(timezone.utc)}: ***** {status} *****") if __name__ == "__main__": diff --git a/tests/retry_test.py b/tests/retry_test.py index 532e6e0..cf94323 100644 --- a/tests/retry_test.py +++ b/tests/retry_test.py @@ -1,10 +1,12 @@ -from trading_ig.rest import IGService, ApiExceededException, TokenInvalidException -import responses -from responses import Response import json + +import pandas as pd +import responses import tenacity +from responses import Response from tenacity import Retrying -import pandas as pd + +from trading_ig.rest import ApiExceededException, IGService, TokenInvalidException RETRYABLE = (ApiExceededException, TokenInvalidException) diff --git a/tests/test_accounts.py b/tests/test_accounts.py index 6427085..98a35e3 100644 --- a/tests/test_accounts.py +++ b/tests/test_accounts.py @@ -1,7 +1,9 @@ -from trading_ig.rest import IGService -import responses import json + import pandas as pd +import responses + +from trading_ig.rest import IGService """ unit tests for accounts methods diff --git a/tests/test_activities.py b/tests/test_activities.py index 5a3e1d6..0cef833 100644 --- a/tests/test_activities.py +++ b/tests/test_activities.py @@ -1,9 +1,11 @@ -from trading_ig.rest import IGService -import responses import json -import pandas as pd -from datetime import datetime, timedelta import re +from datetime import datetime, timedelta, timezone + +import pandas as pd +import responses + +from trading_ig.rest import IGService class TestActivities: @@ -56,7 +58,7 @@ def test_activities_by_date(self): ) ig_service = IGService("username", "password", "api_key", "DEMO") - to_date = datetime.now() + to_date = datetime.now(timezone.utc) from_date = to_date - timedelta(days=7) result = ig_service.fetch_account_activity_by_date(from_date, to_date) diff --git a/tests/test_dealing.py b/tests/test_dealing.py index 219e459..951ce98 100644 --- a/tests/test_dealing.py +++ b/tests/test_dealing.py @@ -1,7 +1,9 @@ -from trading_ig.rest import IGService -import responses import json + import pandas as pd +import responses + +from trading_ig.rest import IGService """ unit tests for dealing methods diff --git a/tests/test_historical_prices.py b/tests/test_historical_prices.py index 0d139e8..38064b7 100644 --- a/tests/test_historical_prices.py +++ b/tests/test_historical_prices.py @@ -1,10 +1,12 @@ -from trading_ig.rest import IGService -import responses +import datetime import json +import re + import pandas as pd -import datetime import pytest -import re +import responses + +from trading_ig.rest import IGService """ unit tests for historical prices methods @@ -125,7 +127,7 @@ def test_historical_prices_v3_num_points_bad_numpoints(self): responses.GET, "https://demo-api.ig.com/gateway/deal/prices/MT.D.GC.Month2.IP", headers={"CST": "abc123", "X-SECURITY-TOKEN": "xyz987"}, - json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, # noqa + json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, status=400, ) @@ -418,7 +420,7 @@ def test_historical_prices_by_epic_and_num_points_bad_numpoints(self): responses.GET, "https://demo-api.ig.com/gateway/deal/prices/MT.D.GC.Month2.IP", headers={"CST": "abc123", "X-SECURITY-TOKEN": "xyz987"}, - json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, # noqa + json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, status=400, ) diff --git a/tests/test_historical_prices_flat.py b/tests/test_historical_prices_flat.py index c1a5add..17edc5b 100644 --- a/tests/test_historical_prices_flat.py +++ b/tests/test_historical_prices_flat.py @@ -1,9 +1,11 @@ -from trading_ig.rest import IGService -import responses +import datetime import json + import pandas as pd -import datetime import pytest +import responses + +from trading_ig.rest import IGService """ unit tests for historical prices methods with flat output formatting @@ -130,7 +132,7 @@ def test_historical_prices_v3_num_points_bad_numpoints(self): responses.GET, "https://demo-api.ig.com/gateway/deal/prices/MT.D.GC.Month2.IP", headers={"CST": "abc123", "X-SECURITY-TOKEN": "xyz987"}, - json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, # noqa + json={"errorCode": "Unable to convert value=3.14159 to type= Integer int"}, status=400, ) diff --git a/tests/test_integration.py b/tests/test_integration.py index 9088172..6c88f24 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,21 +1,23 @@ -from trading_ig.rest import ( - IGService, - IGException, - ApiExceededException, - TokenInvalidException, -) -from trading_ig.config import config -import pandas as pd -from datetime import datetime, timedelta -import pytest -from random import randint, choice import logging import time +from datetime import datetime, timedelta, timezone +from random import choice, randint + +import pandas as pd +import pytest from tenacity import ( Retrying, - wait_exponential, retry_if_exception_type, stop_after_attempt, + wait_exponential, +) + +from trading_ig.config import config +from trading_ig.rest import ( + ApiExceededException, + IGException, + IGService, + TokenInvalidException, ) try: @@ -47,16 +49,6 @@ def limited_retrying(): ) -@pytest.fixture(autouse=True) -def logging_setup(): - """sets logging for each test""" - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s %(levelname)s %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) - - @pytest.fixture(scope="module", params=["2", "3"], ids=["v2 session", "v3 session"]) def ig_service(request, retrying): """ @@ -100,13 +92,16 @@ def watchlist_id(ig_service: IGService): """test fixture creates a dummy watchlist for use in tests, and returns the ID. In teardown it also deletes the dummy watchlist""" epics = ["CS.D.GBPUSD.TODAY.IP", "IX.D.FTSE.DAILY.IP"] - now = datetime.now() + now = datetime.now(timezone.utc) data = ig_service.create_watchlist(f"test_{now.strftime('%Y%m%d%H%H%S')}", epics) watchlist_id = data["watchlistId"] yield watchlist_id ig_service.delete_watchlist(watchlist_id) +logger = logging.getLogger(__name__) + + class TestIntegration: def test_create_session_no_encryption(self, retrying): ig_service = IGService( @@ -163,7 +158,7 @@ def test_fetch_account_activity_by_period(self, ig_service: IGService): assert isinstance(response, pd.DataFrame) def test_fetch_account_activity_by_date(self, ig_service: IGService): - to_date = datetime.now() - timedelta(days=30) + to_date = datetime.now(timezone.utc) - timedelta(days=30) from_date = to_date - timedelta(days=60) response = ig_service.fetch_account_activity_by_date(from_date, to_date) assert isinstance(response, pd.DataFrame) @@ -174,7 +169,7 @@ def test_fetch_account_activity_v2_span(self, ig_service: IGService): assert isinstance(response, pd.DataFrame) def test_fetch_account_activity_v2_dates(self, ig_service): - to_date = datetime.now() - timedelta(days=30) + to_date = datetime.now(timezone.utc) - timedelta(days=30) from_date = to_date - timedelta(days=60) response = ig_service.fetch_account_activity_v2( from_date=from_date, to_date=to_date @@ -182,15 +177,15 @@ def test_fetch_account_activity_v2_dates(self, ig_service): assert isinstance(response, pd.DataFrame) def test_fetch_account_activity_from(self, ig_service: IGService): - to_date = datetime.now() - timedelta(days=30) + to_date = datetime.now(timezone.utc) - timedelta(days=30) from_date = to_date - timedelta(days=60) response = ig_service.fetch_account_activity(from_date=from_date) assert isinstance(response, pd.DataFrame) assert response.shape[1] == 9 def test_fetch_account_activity_from_to(self, ig_service: IGService): - to_date = datetime.now() - timedelta(days=30) - from_date = to_date - timedelta(days=60) + to_date = datetime.now(timezone.utc) - timedelta(days=30) + from_date = to_date - timedelta(days=360) response = ig_service.fetch_account_activity( from_date=from_date, to_date=to_date ) @@ -198,8 +193,8 @@ def test_fetch_account_activity_from_to(self, ig_service: IGService): assert response.shape[1] == 9 def test_fetch_account_activity_detailed(self, ig_service): - to_date = datetime.now() - timedelta(days=30) - from_date = to_date - timedelta(days=60) + to_date = datetime.now(timezone.utc) - timedelta(days=30) + from_date = to_date - timedelta(days=360) response = ig_service.fetch_account_activity( from_date=from_date, to_date=to_date, detailed=True ) @@ -207,7 +202,7 @@ def test_fetch_account_activity_detailed(self, ig_service): assert response.shape[1] == 23 def test_fetch_account_activity_old(self, ig_service: IGService): - from_date = datetime(1970, 1, 1) + from_date = datetime(1970, 1, 1, tzinfo=timezone.utc) to_date = from_date + timedelta(days=60) response = ig_service.fetch_account_activity( from_date=from_date, to_date=to_date @@ -216,7 +211,7 @@ def test_fetch_account_activity_old(self, ig_service: IGService): assert response.shape[0] == 0 def test_fetch_account_activity_fiql(self, ig_service: IGService): - to_date = datetime.now() - timedelta(days=30) + to_date = datetime.now(timezone.utc) - timedelta(days=30) from_date = to_date - timedelta(days=120) response = ig_service.fetch_account_activity( from_date=from_date, to_date=to_date, fiql_filter="channel==PUBLIC_WEB_API" @@ -320,10 +315,10 @@ def test_session_v3_refresh(self, retrying): delay_choice = [(1, 59), (60, 650)] for count in range(1, 20): data = ig_service.fetch_accounts() - logging.info(f"Account count: {len(data)}") + logger.info(f"Account count: {len(data)}") option = choice(delay_choice) wait = randint(option[0], option[1]) - logging.info(f"Waiting for {wait} seconds...") + logger.info(f"Waiting for {wait} seconds...") time.sleep(wait) def test_read_session(self, ig_service: IGService): @@ -787,9 +782,9 @@ def test_create_working_order_guaranteed_stop_loss(self, ig_service: IGService): offer = market_info.snapshot.offer bid = market_info.snapshot.bid - logging.info(f"min bet: {min_bet}") - logging.info(f"offer: {offer}") - logging.info(f"bid: {bid}") + logger.info(f"min bet: {min_bet}") + logger.info(f"offer: {offer}") + logger.info(f"bid: {bid}") if status != "TRADEABLE": pytest.skip("Skipping create working order test, market not open") @@ -810,7 +805,7 @@ def test_create_working_order_guaranteed_stop_loss(self, ig_service: IGService): stop_level=None, ) - logging.info( + logger.info( f"result: {create_result['dealStatus']}, reason {create_result['reason']}" ) @@ -833,7 +828,7 @@ def test_create_working_order_guaranteed_stop_loss(self, ig_service: IGService): deal_id=create_result["dealId"], ) - logging.info( + logger.info( f"result: {update_result['dealStatus']}, reason {update_result['reason']}" ) diff --git a/tests/test_positions.py b/tests/test_positions.py index e746992..8c408ff 100644 --- a/tests/test_positions.py +++ b/tests/test_positions.py @@ -1,7 +1,9 @@ -from trading_ig.rest import IGService -import responses import json + import pandas as pd +import responses + +from trading_ig.rest import IGService class TestPositions: diff --git a/tests/test_rate_limiter.py b/tests/test_rate_limiter.py index 8dc88a8..47505c9 100644 --- a/tests/test_rate_limiter.py +++ b/tests/test_rate_limiter.py @@ -1,8 +1,10 @@ -from trading_ig.rest import IGService -import responses import json import time +import responses + +from trading_ig.rest import IGService + """ unit tests for rate limiter """ diff --git a/tests/test_session.py b/tests/test_session.py index 71fe6f9..cac7a0d 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -1,7 +1,9 @@ -from trading_ig.rest import IGService, IGException -import responses import json + import pytest +import responses + +from trading_ig.rest import IGException, IGService """ unit tests for session methods @@ -75,7 +77,7 @@ def test_login_v1_encrypted_happy(self): responses.GET, "https://demo-api.ig.com/gateway/deal/session/encryptionKey", json={ - "encryptionKey": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAp9te7zwed8HhdRFsn47EI8exZ1Yi+bJoKtclGTiuaP1T+4AclNqB2mIya/Ik6IV6A2pt4FFVoqvrhJA46dWi4XgA4Ojhl2Xxw4++blAMgT3jU7N5nY13LdJzZuYv/oPZKRcEj6RrlBV68HjrTnjAMWARl0jFbVCiLWovTGJ0stx/zJAKX0GFyuUlsoaJISJJRYeOLUtZ8Z4BE6ZkmKnz4V8YNyyoWCyXQp+IKCZrfoEdlMOPBgsjbRy02Gh9xZqcm2erLsp40F+w3AjHUqQQi7eQuPQaPWq9Lhm8cVDH2CB2BtfM8Ew8T5/A36eqa5eoeQcZaMnLUQP5UYtG2Wd//wIDAQAB", # noqa + "encryptionKey": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAp9te7zwed8HhdRFsn47EI8exZ1Yi+bJoKtclGTiuaP1T+4AclNqB2mIya/Ik6IV6A2pt4FFVoqvrhJA46dWi4XgA4Ojhl2Xxw4++blAMgT3jU7N5nY13LdJzZuYv/oPZKRcEj6RrlBV68HjrTnjAMWARl0jFbVCiLWovTGJ0stx/zJAKX0GFyuUlsoaJISJJRYeOLUtZ8Z4BE6ZkmKnz4V8YNyyoWCyXQp+IKCZrfoEdlMOPBgsjbRy02Gh9xZqcm2erLsp40F+w3AjHUqQQi7eQuPQaPWq9Lhm8cVDH2CB2BtfM8Ew8T5/A36eqa5eoeQcZaMnLUQP5UYtG2Wd//wIDAQAB", "timeStamp": "1601218928621", }, status=200, @@ -180,7 +182,7 @@ def test_login_v2_encrypted_happy(self): responses.GET, "https://demo-api.ig.com/gateway/deal/session/encryptionKey", json={ - "encryptionKey": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAp9te7zwed8HhdRFsn47EI8exZ1Yi+bJoKtclGTiuaP1T+4AclNqB2mIya/Ik6IV6A2pt4FFVoqvrhJA46dWi4XgA4Ojhl2Xxw4++blAMgT3jU7N5nY13LdJzZuYv/oPZKRcEj6RrlBV68HjrTnjAMWARl0jFbVCiLWovTGJ0stx/zJAKX0GFyuUlsoaJISJJRYeOLUtZ8Z4BE6ZkmKnz4V8YNyyoWCyXQp+IKCZrfoEdlMOPBgsjbRy02Gh9xZqcm2erLsp40F+w3AjHUqQQi7eQuPQaPWq9Lhm8cVDH2CB2BtfM8Ew8T5/A36eqa5eoeQcZaMnLUQP5UYtG2Wd//wIDAQAB", # noqa + "encryptionKey": "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAp9te7zwed8HhdRFsn47EI8exZ1Yi+bJoKtclGTiuaP1T+4AclNqB2mIya/Ik6IV6A2pt4FFVoqvrhJA46dWi4XgA4Ojhl2Xxw4++blAMgT3jU7N5nY13LdJzZuYv/oPZKRcEj6RrlBV68HjrTnjAMWARl0jFbVCiLWovTGJ0stx/zJAKX0GFyuUlsoaJISJJRYeOLUtZ8Z4BE6ZkmKnz4V8YNyyoWCyXQp+IKCZrfoEdlMOPBgsjbRy02Gh9xZqcm2erLsp40F+w3AjHUqQQi7eQuPQaPWq9Lhm8cVDH2CB2BtfM8Ew8T5/A36eqa5eoeQcZaMnLUQP5UYtG2Wd//wIDAQAB", "timeStamp": "1601218928621", }, status=200, diff --git a/trading_ig/__init__.py b/trading_ig/__init__.py index 63bdf8d..5ca31ba 100644 --- a/trading_ig/__init__.py +++ b/trading_ig/__init__.py @@ -1,15 +1,9 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - """ IG Markets API Library for Python https://github.com/ig-python/trading-ig/ by Femto Trader - https://github.com/femtotrader """ -from __future__ import absolute_import, division, print_function - - from .rest import IGService from .stream import IGStreamService diff --git a/trading_ig/config.py b/trading_ig/config.py index f01dbe1..c97fe2f 100644 --- a/trading_ig/config.py +++ b/trading_ig/config.py @@ -1,8 +1,5 @@ -#!/usr/bin/env python -# -*- coding:utf-8 -*- - -import os import logging +import os ENV_VAR_ROOT = "IG_SERVICE" CONFIG_FILE_NAME = "trading_ig_config.py" @@ -10,7 +7,7 @@ logger = logging.getLogger(__name__) -class ConfigEnvVar(object): +class ConfigEnvVar: def __init__(self, env_var_base): self.ENV_VAR_BASE = env_var_base @@ -26,22 +23,21 @@ def __getattr__(self, key): try: return os.environ[env_var] except KeyError: - raise Exception("Environment variable '%s' doesn't exist" % env_var) + raise Exception(f"Environment variable '{env_var}' doesn't exist") try: from trading_ig_config import config - logger.info("import config from %s" % CONFIG_FILE_NAME) + logger.info(f"import config from {CONFIG_FILE_NAME}") except Exception: logger.warning("can't import config from config file") try: config = ConfigEnvVar(ENV_VAR_ROOT) - logger.info("import config from environment variables '%s_...'" % ENV_VAR_ROOT) + logger.info(f"import config from environment variables '{ENV_VAR_ROOT}_...'") except Exception: logger.warning("can't import config from environment variables") raise ( - """Can't import config - you might create a '%s' filename or use -environment variables such as '%s_...'""" - % (CONFIG_FILE_NAME, ENV_VAR_ROOT) + f"Can't import config - you might create a '{CONFIG_FILE_NAME}' " + f"filename or use environment variables such as '{ENV_VAR_ROOT}_...'" ) diff --git a/trading_ig/lightstreamer.py b/trading_ig/lightstreamer.py deleted file mode 100644 index 0d6be9c..0000000 --- a/trading_ig/lightstreamer.py +++ /dev/null @@ -1,467 +0,0 @@ -#!/usr/bin/python3 - -# Copyright (c) Lightstreamer Srl. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import logging -import threading -import traceback -import sys -import warnings - -from six.moves.urllib.request import urlopen as _urlopen -from six.moves.urllib.parse import urlparse as parse_url, urljoin, urlencode - -try: - from systemd.daemon import notify -except ImportError: - notify = None - - -def _url_encode(params): - return urlencode(params).encode("utf-8") - - -def _iteritems(d): - return iter(d.items()) - - -CONNECTION_URL_PATH = "lightstreamer/create_session.txt" -BIND_URL_PATH = "lightstreamer/bind_session.txt" -CONTROL_URL_PATH = "lightstreamer/control.txt" -# Request parameter to create and activate a new Table. -OP_ADD = "add" -# Request parameter to delete a previously created Table. -OP_DELETE = "delete" -# Request parameter to force closure of an existing session. -OP_DESTROY = "destroy" -# List of possible server responses -PROBE_CMD = "PROBE" -END_CMD = "END" -LOOP_CMD = "LOOP" -ERROR_CMD = "ERROR" -SYNC_ERROR_CMD = "SYNC ERROR" -OK_CMD = "OK" - -log = logging.getLogger(__name__) - - -class Subscription(object): - """Represents a Subscription to be submitted to a Lightstreamer Server.""" - - def __init__(self, mode, items, fields, adapter=""): - warnings.warn( - "trading_ig.lightstreamer.Subscription is deprecated, and will be removed " - "in a future version; use the official Lightstreamer Python client instead", - DeprecationWarning, - 2, - ) - self.item_names = items - self._items_map = {} - self.field_names = fields - self.adapter = adapter - self.mode = mode - self.snapshot = "true" - self._listeners = [] - - def _decode(self, value, last): - """Decode the field value according to - Lightstremar Text Protocol specifications. - """ - if value == "$": - return "" - elif value == "#": - return None - elif not value: - return last - elif value[0] in "#$": - value = value[1:] - - return value - - def addlistener(self, listener): - self._listeners.append(listener) - - def notifyupdate(self, item_line): - """Invoked by LSClient each time Lightstreamer Server pushes - a new item event. - """ - # Tokenize the item line as sent by Lightstreamer - toks = item_line.rstrip("\r\n").split("|") - undecoded_item = dict(list(zip(self.field_names, toks[1:]))) - - # Retrieve the previous item stored into the map, if present. - # Otherwise create a new empty dict. - item_pos = int(toks[0]) - curr_item = self._items_map.get(item_pos, {}) - # Update the map with new values, merging with the - # previous ones if any. - self._items_map[item_pos] = dict( - [ - (k, self._decode(v, curr_item.get(k))) - for k, v in list(undecoded_item.items()) - ] - ) - # Make an item info as a new event to be passed to listeners - item_info = { - "pos": item_pos, - "name": self.item_names[item_pos - 1], - "values": self._items_map[item_pos], - } - - # Update each registered listener with new event - for on_item_update in self._listeners: - on_item_update(item_info) - - -class LSClient(object): - """Manages the communication with Lightstreamer Server""" - - def __init__(self, base_url, adapter_set="", user="", password=""): - warnings.warn( - "trading_ig.lightstreamer.LSClient is deprecated, and will be removed in " - "a future version; use the official Lightstreamer Python client instead", - DeprecationWarning, - 2, - ) - self._base_url = parse_url(base_url) - self._adapter_set = adapter_set - self._user = user - self._password = password - self._session = {} - self._subscriptions = {} - self._current_subscription_key = 0 - self._stream_connection = None - self._stream_connection_thread = None - self._bind_counter = 0 - self.content_length = 1000000000 - - def _encode_params(self, params): - """Encode the parameter for HTTP POST submissions, but - only for non empty values...""" - return _url_encode(dict([(k, v) for (k, v) in _iteritems(params) if v])) - - def _call(self, base_url, url, body): - """Open a network connection and performs HTTP Post - with provided body. - """ - # Combines the "base_url" with the - # required "url" to be used for the specific request. - url = urljoin(base_url.geturl(), url) - return _urlopen(url, data=self._encode_params(body)) - - def _set_control_link_url(self, custom_address=None): - """Set the address to use for the Control Connection - in such cases where Lightstreamer is behind a Load Balancer. - """ - if custom_address is None: - self._control_url = self._base_url - else: - parsed_custom_address = parse_url("//" + custom_address) - self._control_url = parsed_custom_address._replace(scheme=self._base_url[0]) - - def _control(self, params): - """Create a Control Connection to send control commands - that manage the content of Stream Connection. - """ - params["LS_session"] = self._session["SessionId"] - response = self._call(self._control_url, CONTROL_URL_PATH, params) - return response.readline().decode("utf-8").rstrip() - - def _read_from_stream(self): - """Read a single line of content of the Stream Connection.""" - line = self._stream_connection.readline().decode("utf-8").rstrip() - return line - - def connect(self): - """Establish a connection to Lightstreamer Server to create - a new session. - """ - - if not notify and sys.platform.startswith("linux"): - log.warning( - "systemd.daemon not available, no watchdog notifications will be sent." - ) - - self._stream_connection = self._call( - self._base_url, - CONNECTION_URL_PATH, - { - "LS_op2": "create", - "LS_cid": "mgQkwtwdysogQz2BJ4Ji kOj2Bg", - "LS_adapter_set": self._adapter_set, - "LS_user": self._user, - "LS_password": self._password, - "LS_content_length": self.content_length, - }, - ) - stream_line = self._read_from_stream() - self._handle_stream(stream_line) - - def bind(self): - """Replace a completely consumed connection in listening for an active - Session. - """ - self._stream_connection = self._call( - self._control_url, - BIND_URL_PATH, - { - "LS_session": self._session["SessionId"], - "LS_content_length": self.content_length, - }, - ) - - self._bind_counter += 1 - stream_line = self._read_from_stream() - self._handle_stream(stream_line) - - def _handle_stream(self, stream_line): - if stream_line == OK_CMD: - # Parsing session inkion - while 1: - next_stream_line = self._read_from_stream() - if next_stream_line: - session_key, session_value = next_stream_line.split(":", 1) - self._session[session_key] = session_value - else: - break - - # Setup of the control link url - self._set_control_link_url(self._session.get("ControlAddress")) - - # Start a new thread to handle real time updates sent - # by Lightstreamer Server on the stream connection. - self._stream_connection_thread = threading.Thread( - name="STREAM-CONN-THREAD-{0}".format(self._bind_counter), - target=self._receive, - ) - self._stream_connection_thread.setDaemon(True) - # Add "active connection" attribute to running thread - setattr(self._stream_connection_thread, "active_connection", True) - self._stream_connection_thread.start() - else: - lines = self._stream_connection.readlines() - lines.insert(0, stream_line) - log.error("Server response error: \n{0}".format("".join(lines))) - raise IOError() - - def _join(self): - """Await the natural STREAM-CONN-THREAD termination.""" - if self._stream_connection_thread: - log.debug("Waiting for thread to terminate") - self._stream_connection_thread.active_connection = False - self._stream_connection_thread.join() - self._stream_connection_thread = None - log.debug("Thread terminated") - - def disconnect(self): - """Request to close the session previously opened with - the connect() invocation. - """ - if self._stream_connection is not None: - # Exits stream thread loop, joins and exits stream thread, closes connection - self._join() - log.debug("Connection closed") - print("DISCONNECTED FROM LIGHTSTREAMER") - else: - log.warning("No connection to Lightstreamer") - - def destroy(self): - """Destroy the session previously opened with - the connect() invocation. - """ - if self._stream_connection is not None: - server_response = self._control({"LS_op": OP_DESTROY}) - if server_response == OK_CMD: - # There is no need to explicitly close the connection, - # since it is handled by thread completion. - self._join() - else: - log.warning("No connection to Lightstreamer") - - def subscribe(self, subscription): - """ "Perform a subscription request to Lightstreamer Server.""" - # Register the Subscription with a new subscription key - self._current_subscription_key += 1 - self._subscriptions[self._current_subscription_key] = subscription - - # Send the control request to perform the subscription - server_response = self._control( - { - "LS_Table": self._current_subscription_key, - "LS_op": OP_ADD, - "LS_data_adapter": subscription.adapter, - "LS_mode": subscription.mode, - "LS_schema": " ".join(subscription.field_names), - "LS_id": " ".join(subscription.item_names), - } - ) - log.debug("Server response ---> <{0}>".format(server_response)) - return self._current_subscription_key - - def unsubscribe(self, subcription_key): - """Unregister the Subscription associated to the - specified subscription_key. - """ - if subcription_key in self._subscriptions: - server_response = self._control( - {"LS_Table": subcription_key, "LS_op": OP_DELETE} - ) - log.debug("Server response ---> <{0}>".format(server_response)) - - if server_response == OK_CMD: - del self._subscriptions[subcription_key] - log.info("Unsubscribed successfully") - else: - log.warning("Server error") - else: - log.warning("No subscription key {0} found!".format(subcription_key)) - - def _forward_update_message(self, update_message): - """Forwards the real time update to the relative - Subscription instance for further dispatching to its listeners. - """ - log.debug("Received update message ---> <{0}>".format(update_message)) - tok = update_message.split(",", 1) - table, item = int(tok[0]), tok[1] - if table in self._subscriptions: - self._subscriptions[table].notifyupdate(item) - else: - log.warning("No subscription found!") - - def _receive(self): - rebind = False - receive = True - while receive and self._stream_connection_thread.active_connection: - log.debug("Waiting for a new message") - try: - message = self._read_from_stream() - log.debug("Received message ---> <{0}>".format(message)) - except Exception: - log.error("Communication error") - print(traceback.format_exc()) - message = None - - if notify: - notify("WATCHDOG=1") - - if message is None: - receive = False - log.warning("No new message received") - elif message == PROBE_CMD: - # Skipping the PROBE message, keep on receiving messages. - log.debug("PROBE message") - elif message.startswith(ERROR_CMD): - # Terminate the receiving loop on ERROR message - receive = False - log.error("ERROR") - elif message.startswith(LOOP_CMD): - # Terminate the the receiving loop on LOOP message. - # A complete implementation should proceed with - # a rebind of the session. - log.debug("LOOP") - rebind = True - receive = False - elif message.startswith(SYNC_ERROR_CMD): - # Terminate the receiving loop on SYNC ERROR message. - # A complete implementation should create a new session - # and re-subscribe to all the old items and relative fields. - log.error("SYNC ERROR") - receive = False - elif message.startswith(END_CMD): - # Terminate the receiving loop on END message. - # The session has been forcibly closed on the server side. - # A complete implementation should handle the - # "cause_code" if present. - log.info("Connection closed by the server") - receive = False - elif message.startswith("Preamble"): - # Skipping Preamble message, keep on receiving messages. - log.debug("Preamble") - else: - self._forward_update_message(message) - - if not rebind: - log.debug("Closing connection") - # Clear internal data structures for session - # and subscriptions management. - self._stream_connection.close() - self._stream_connection = None - self._session.clear() - self._subscriptions.clear() - self._current_subscription_key = 0 - else: - log.debug("Binding to this active session") - self._stream_connection = None - self.bind() - - -if __name__ == "__main__": - logging.basicConfig(level=logging.INFO) - - # Establishing a new connection to Lightstreamer Server - print("Starting connection") - # lightstreamer_client = LSClient("http://localhost:8080", "DEMO") - lightstreamer_client = LSClient("http://push.lightstreamer.com", "DEMO") - try: - lightstreamer_client.connect() - except Exception: - print("Unable to connect to Lightstreamer Server") - print(traceback.format_exc()) - - sys.exit(1) - - # Making a new Subscription in MERGE mode - subscription = Subscription( - mode="MERGE", - items=[ - "item1", - "item2", - "item3", - "item4", - "item5", - "item6", - "item7", - "item8", - "item9", - "item10", - "item11", - "item12", - ], - fields=["stock_name", "last_price", "time", "bid", "ask"], - adapter="QUOTE_ADAPTER", - ) - - # A simple function acting as a Subscription listener - def on_item_update(item_update): - print( - "{stock_name:<19}: Last{last_price:>6} - Time {time:<8} - " - "Bid {bid:>5} - Ask {ask:>5}".format(**item_update["values"]) - ) - - # Adding the "on_item_update" function to Subscription - subscription.addlistener(on_item_update) - - # Registering the Subscription - sub_key = lightstreamer_client.subscribe(subscription) - - def handler(): - # Unsubscribing from Lightstreamer by using the subscription key - lightstreamer_client.unsubscribe(sub_key) - # Disconnecting - lightstreamer_client.disconnect() - - import atexit - - atexit.register(handler) diff --git a/trading_ig/rest.py b/trading_ig/rest.py index 692c3e4..6bf65e9 100644 --- a/trading_ig/rest.py +++ b/trading_ig/rest.py @@ -1,31 +1,29 @@ -#!/usr/bin/env python -# -*- coding:utf-8 -*- - """ IG Markets REST API Library for Python https://labs.ig.com/rest-trading-api-reference Original version by Lewis Barber - 2014 - https://uk.linkedin.com/in/lewisbarber/ Modified by Femto Trader - 2014-2015 - https://github.com/femtotrader/ -""" # noqa +""" import json import logging import time -from base64 import b64encode, b64decode +from base64 import b64decode, b64encode +from datetime import datetime, timedelta, timezone +from urllib.parse import parse_qs, urlparse from Crypto.Cipher import PKCS1_v1_5 from Crypto.PublicKey import RSA from requests import Session -from urllib.parse import urlparse, parse_qs -from datetime import timedelta, datetime -from .utils import _HAS_PANDAS, _HAS_MUNCH from .utils import ( - conv_resol, - conv_datetime, - conv_to_ms, + _HAS_MUNCH, + _HAS_PANDAS, DATE_FORMATS, api_limit_hit, + conv_datetime, + conv_resol, + conv_to_ms, token_invalid, ) @@ -33,26 +31,28 @@ from .utils import munchify if _HAS_PANDAS: - from .utils import pd from pandas import json_normalize + from .utils import pd + +from queue import Empty, Queue from threading import Thread -from queue import Queue, Empty logger = logging.getLogger(__name__) +D_BASE_URL = { + "live": "https://api.ig.com/gateway/deal", + "demo": "https://demo-api.ig.com/gateway/deal", +} + class ApiExceededException(Exception): """Raised when our code hits the IG endpoint too often""" - pass - class TokenInvalidException(Exception): """Raised when the session token is invalid or expired""" - pass - class IGException(Exception): pass @@ -61,10 +61,8 @@ class IGException(Exception): class KycRequiredException(Exception): """Raised when IG needs the user to confirm or re-confirm their KYC status""" - pass - -class IGSessionCRUD(object): +class IGSessionCRUD: """Session with CRUD operation""" BASE_URL = None @@ -92,8 +90,7 @@ def _get_session(self, session): """ if session is None: session = self.session # requests Session - else: - session = session + return session def _url(self, endpoint): @@ -164,11 +161,6 @@ def req(self, action, endpoint, params, session, version): class IGService: - D_BASE_URL = { - "live": "https://api.ig.com/gateway/deal", - "demo": "https://demo-api.ig.com/gateway/deal", - } - API_KEY = None IG_USERNAME = None IG_PASSWORD = None @@ -198,10 +190,10 @@ def __init__( self._use_rate_limiter = use_rate_limiter self._bucket_threads_run = False try: - self.BASE_URL = self.D_BASE_URL[acc_type.lower()] + self.BASE_URL = D_BASE_URL[acc_type.lower()] except Exception: raise IGException( - "Invalid account type '%s', please provide LIVE or DEMO" % acc_type + f"Invalid account type '{acc_type}', please provide LIVE or DEMO" ) self.return_dataframe = return_dataframe @@ -280,17 +272,12 @@ def setup_rate_limiter( token_bucket_non_trading_thread.start() self._non_trading_times = [] - # TODO - # Create a leaky token bucket for allowanceAccountHistoricalData - return - def _token_bucket_trading( self, ): while self._bucket_threads_run: time.sleep(60.0 / self._trading_requests_per_minute) self._trading_requests_queue.put(True, block=True) - return def _token_bucket_non_trading( self, @@ -298,7 +285,6 @@ def _token_bucket_non_trading( while self._bucket_threads_run: time.sleep(60.0 / self._non_trading_requests_per_minute) self._non_trading_requests_queue.put(True, block=True) - return def trading_rate_limit_pause_or_pass( self, @@ -315,7 +301,6 @@ def trading_rate_limit_pause_or_pass( f"Number of trading requests in last 60 seconds = " f"{len(self._trading_times)} of {self._trading_requests_per_minute}" ) - return def non_trading_rate_limit_pause_or_pass( self, @@ -333,23 +318,20 @@ def non_trading_rate_limit_pause_or_pass( f"{len(self._non_trading_times)} of " f"{self._non_trading_requests_per_minute}" ) - return def _exit_bucket_threads( self, ): - if self._use_rate_limiter: - if self._bucket_threads_run: - self._bucket_threads_run = False - try: - self._trading_requests_queue.get(block=False) - except Empty: - pass - try: - self._non_trading_requests_queue.get(block=False) - except Empty: - pass - return + if self._use_rate_limiter and self._bucket_threads_run: + self._bucket_threads_run = False + try: + self._trading_requests_queue.get(block=False) + except Empty: + pass + try: + self._non_trading_requests_queue.get(block=False) + except Empty: + pass def _get_session(self, session): """Returns a Requests session (from self.session) if session is None @@ -360,10 +342,8 @@ def _get_session(self, session): session = self.session # requests Session else: assert isinstance(session, Session), ( - "session must be not %s" - % type(session) + f"session must be not {type(session)}" ) - session = session return session def _req(self, action, endpoint, params, session, version="1", check=True): @@ -399,7 +379,7 @@ def _request(self, action, endpoint, params, session, version="1", check=True): raise ApiExceededException() if token_invalid(response.text): logger.warning("Invalid session token, triggering refresh...") - self._valid_until = datetime.now() - timedelta(seconds=15) + self._valid_until = datetime.now(timezone.utc) - timedelta(seconds=15) raise TokenInvalidException() return response @@ -443,7 +423,7 @@ def expand_columns(data, d_cols, flag_col_prefix=False, col_overlap_allowed=None colname = col data[colname] = ser.map(lambda x: x[col], na_action="ignore") else: - raise (NotImplementedError("col overlap: %r" % col)) + raise (NotImplementedError(f"col overlap: {col}")) return data # -------- END ------- # @@ -618,9 +598,9 @@ def fetch_account_activity_by_date( def fetch_account_activity_v2( self, - from_date: datetime = None, - to_date: datetime = None, - max_span_seconds: int = None, + from_date: datetime | None = None, + to_date: datetime | None = None, + max_span_seconds: int | None = None, page_size: int = 20, session=None, ): @@ -685,11 +665,11 @@ def fetch_account_activity_v2( def fetch_account_activity( self, - from_date: datetime = None, - to_date: datetime = None, + from_date: datetime | None = None, + to_date: datetime | None = None, detailed=False, - deal_id: str = None, - fiql_filter: str = None, + deal_id: str | None = None, + fiql_filter: str | None = None, page_size: int = 50, session=None, ): @@ -751,11 +731,13 @@ def fetch_account_activity( query = parse_qs(parse_result.query) logger.debug(f"fetch_account_activity() next query: '{query}'") if "from" in query: - params["from"] = query["from"][0] + # from_str = query["from"][0] + params["from"] = query["from"][0][:19] else: del params["from"] if "to" in query: - params["to"] = query["to"][0] + # to_str = query["from"][0] + params["to"] = query["from"][0][:19] else: del params["to"] @@ -934,8 +916,8 @@ def fetch_deal_by_deal_reference(self, deal_reference, session=None): action = "read" for i in range(5): response = self._req(action, endpoint, params, session, version) - if not response.status_code == 200: - logger.info("Deal reference %s not found, retrying." % deal_reference) + if response.status_code != 200: + logger.info(f"Deal reference {deal_reference} not found, retrying.") time.sleep(1) else: break @@ -952,8 +934,8 @@ def fetch_open_position_by_deal_id(self, deal_id, session=None): action = "read" for i in range(5): response = self._req(action, endpoint, params, session, version) - if not response.status_code == 200: - logger.info("Deal id %s not found, retrying." % deal_id) + if response.status_code != 200: + logger.info(f"Deal id {deal_id} not found, retrying.") time.sleep(1) else: break @@ -976,7 +958,7 @@ def fetch_open_positions(self, session=None, version="2"): action = "read" for i in range(5): response = self._req(action, endpoint, params, session, version) - if not response.status_code == 200: + if response.status_code != 200: logger.info("Error fetching open positions, retrying.") time.sleep(1) else: @@ -1406,7 +1388,7 @@ def fetch_repeat_dealing_window(self, epic=None, session=None): action = "read" for i in range(5): response = self._req(action, endpoint, params, session, version) - if not response.status_code == 200: + if response.status_code != 200: logger.info("Error fetching repeat dealing window, retrying.") time.sleep(1) else: @@ -1603,10 +1585,10 @@ def format_prices(self, prices, version, flag_calc_spread=False): def cols(typ): return { - "openPrice.%s" % typ: "Open", - "highPrice.%s" % typ: "High", - "lowPrice.%s" % typ: "Low", - "closePrice.%s" % typ: "Close", + f"openPrice.{typ}": "Open", + f"highPrice.{typ}": "High", + f"lowPrice.{typ}": "Low", + f"closePrice.{typ}": "Close", "lastTradedVolume": "Volume", } @@ -1935,10 +1917,12 @@ def fetch_historical_prices_by_epic_and_date_range( def log_allowance(self, data): remaining_allowance = data["allowance"]["remainingAllowance"] allowance_expiry_secs = data["allowance"]["allowanceExpiry"] - allowance_expiry = datetime.today() + timedelta(seconds=allowance_expiry_secs) + allowance_expiry = datetime.now(timezone.utc) + timedelta( + seconds=allowance_expiry_secs + ) logger.info( - "Historic price data allowance: %s remaining until %s" - % (remaining_allowance, allowance_expiry) + f"Historic price data allowance: {remaining_allowance} " + f"remaining until {allowance_expiry}" ) # -------- END -------- # @@ -2135,7 +2119,7 @@ def _handle_oauth(self, oauth): self.session.headers.update({"Authorization": f"{token_type} {access_token}"}) self._refresh_token = oauth["refresh_token"] validity = int(oauth["expires_in"]) - self._valid_until = datetime.now() + timedelta(seconds=validity) + self._valid_until = datetime.now(timezone.utc) + timedelta(seconds=validity) def _check_session(self): """ @@ -2145,18 +2129,21 @@ def _check_session(self): - if not, a new session will be created """ logger.debug("Checking session status...") - if self._valid_until is not None and datetime.now() > self._valid_until: - if self._refresh_token: - # we are in a v3 session, need to refresh - try: - logger.info("Current session has expired, refreshing...") - self.refresh_session() - except IGException: - logger.info("Refresh failed, logging in again...") - self._refresh_token = None - self._valid_until = None - del self.session.headers["Authorization"] - self.create_session(version="3") + if ( + self._valid_until is not None + and datetime.now(timezone.utc) > self._valid_until + and self._refresh_token + ): + # we are in a v3 session, need to refresh + try: + logger.info("Current session has expired, refreshing...") + self.refresh_session() + except IGException: + logger.info("Refresh failed, logging in again...") + self._refresh_token = None + self._valid_until = None + del self.session.headers["Authorization"] + self.create_session(version="3") def switch_account(self, account_id, default_account, session=None): """Switches active accounts, optionally setting the default account""" @@ -2177,7 +2164,7 @@ def read_session(self, fetch_session_tokens="false", session=None): action = "read" response = self._req(action, endpoint, params, session, version) if not response.ok: - raise IGException("Error in read_session() %s" % response.status_code) + raise IGException(f"Error in read_session() {response.status_code}") data = self.parse_response(response.text) return data diff --git a/trading_ig/stream.py b/trading_ig/stream.py index 1cb0b98..a681f4a 100644 --- a/trading_ig/stream.py +++ b/trading_ig/stream.py @@ -1,18 +1,13 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -from __future__ import absolute_import, division, print_function - +import logging import sys import traceback -import logging -from lightstreamer.client import LightstreamerClient, Subscription, ClientListener +from lightstreamer.client import ClientListener, LightstreamerClient, Subscription logger = logging.getLogger(__name__) -class IGStreamService(object): +class IGStreamService: def __init__(self, ig_service): self.ig_service = ig_service self.lightstreamerEndpoint = None @@ -29,10 +24,10 @@ def create_session(self, encryption=False, version="2"): self.lightstreamerEndpoint = ig_session["lightstreamerEndpoint"] cst = self.ig_service.session.headers["CST"] xsecuritytoken = self.ig_service.session.headers["X-SECURITY-TOKEN"] - ls_password = "CST-%s|XST-%s" % (cst, xsecuritytoken) + ls_password = f"CST-{cst}|XST-{xsecuritytoken}" # Establishing a new connection to Lightstreamer Server - logger.info("Starting connection with %s" % self.lightstreamerEndpoint) + logger.info(f"Starting connection with {self.lightstreamerEndpoint}") self.ls_client = LightstreamerClient(self.lightstreamerEndpoint, None) self.ls_client.connectionDetails.setUser(self.acc_number) self.ls_client.connectionDetails.setPassword(ls_password) diff --git a/trading_ig/streamer/manager.py b/trading_ig/streamer/manager.py index 0cff441..6c8c06a 100644 --- a/trading_ig/streamer/manager.py +++ b/trading_ig/streamer/manager.py @@ -1,13 +1,13 @@ import logging +import time from queue import Queue from threading import Thread -import time -from lightstreamer.client import SubscriptionListener, ItemUpdate +from lightstreamer.client import ItemUpdate, SubscriptionListener from trading_ig import IGStreamService -from .ticker import Ticker -from .ticker import TickerSubscription + +from .ticker import Ticker, TickerSubscription logger = logging.getLogger(__name__) diff --git a/trading_ig/streamer/objects.py b/trading_ig/streamer/objects.py index a8a3101..0febfbf 100644 --- a/trading_ig/streamer/objects.py +++ b/trading_ig/streamer/objects.py @@ -1,5 +1,4 @@ -from datetime import datetime - +from datetime import datetime, timezone nan = float("nan") @@ -17,7 +16,9 @@ def set_timestamp_by_name(self, attr_name, values, key): try: if key in values: setattr( - self, attr_name, datetime.fromtimestamp(int(values[key]) / 1000) + self, + attr_name, + datetime.fromtimestamp(int(values[key]) / 1000, tz=timezone.utc), ) except TypeError: # ignore, there will be plenty of dud values diff --git a/trading_ig/streamer/ticker.py b/trading_ig/streamer/ticker.py index 44b0ad3..a37a5cc 100644 --- a/trading_ig/streamer/ticker.py +++ b/trading_ig/streamer/ticker.py @@ -1,31 +1,33 @@ from dataclasses import dataclass from datetime import datetime + from lightstreamer.client import Subscription -from .objects import nan, StreamObject + +from .objects import StreamObject, nan + +TICKER_FIELDS = [ + "BID", + "OFR", + "LTP", + "LTV", + "TTV", + "UTM", + "DAY_OPEN_MID", + "DAY_NET_CHG_MID", + "DAY_PERC_CHG_MID", + "DAY_HIGH", + "DAY_LOW", +] class TickerSubscription(Subscription): """Represents a subscription for tick prices""" - TICKER_FIELDS = [ - "BID", - "OFR", - "LTP", - "LTV", - "TTV", - "UTM", - "DAY_OPEN_MID", - "DAY_NET_CHG_MID", - "DAY_PERC_CHG_MID", - "DAY_HIGH", - "DAY_LOW", - ] - def __init__(self, epic: str): super().__init__( mode="DISTINCT", items=[f"CHART:{epic}:TICK"], - fields=self.TICKER_FIELDS, + fields=TICKER_FIELDS, ) def __repr__(self) -> str: diff --git a/trading_ig/utils.py b/trading_ig/utils.py index 5cc25e2..90e5839 100644 --- a/trading_ig/utils.py +++ b/trading_ig/utils.py @@ -1,9 +1,7 @@ -#!/usr/bin/env python -# -*- coding:utf-8 -*- - -import os import logging +import os import traceback + import six logger = logging.getLogger(__name__) @@ -12,8 +10,8 @@ OPT_URL = "https://trading-ig.readthedocs.io/en/latest/faq.html#optional-dependencies" try: - import pandas as pd import numpy as np # noqa + import pandas as pd except ImportError: _HAS_PANDAS = False logger.warning(f"pandas is not present in the environment. See {OPT_URL}") @@ -59,7 +57,7 @@ def conv_resol(resolution): return d[offset] else: logger.error(traceback.format_exc()) - logger.warning("conv_resol returns '%s'" % resolution) + logger.warning(f"conv_resol returns '{resolution}'") return resolution else: return resolution @@ -72,14 +70,13 @@ def conv_datetime(dt, version=2): version 3 = 2014/12/15 00:00:00 """ try: - if isinstance(dt, six.string_types): - if _HAS_PANDAS: - dt = pd.to_datetime(dt) + if isinstance(dt, six.string_types) and _HAS_PANDAS: + dt = pd.to_datetime(dt) fmt = DATE_FORMATS[int(version)] return dt.strftime(fmt) except (ValueError, TypeError): - logger.warning("conv_datetime returns %s" % dt) + logger.warning(f"conv_datetime returns {dt}") return dt @@ -92,18 +89,18 @@ def conv_to_ms(td): return int(td.total_seconds() * 1000.0) except ValueError: logger.error(traceback.format_exc()) - logger.warning("conv_to_ms returns '%s'" % td) + logger.warning(f"conv_to_ms returns '{td}'") return td def remove(cache): """Remove cache""" try: - filename = "%s.sqlite" % cache - print("remove %s" % filename) + filename = f"{cache}.sqlite" + print(f"remove {filename}") os.remove(filename) - except Exception: - pass + except Exception as ex: + logger.error(f"Problem removing cache file: {ex}") def print_full(x):