diff options
author | Ismaël Bouya <ismael.bouya@normalesup.org> | 2018-03-09 19:18:02 +0100 |
---|---|---|
committer | Ismaël Bouya <ismael.bouya@normalesup.org> | 2018-03-09 19:18:02 +0100 |
commit | 34eb08f759a440af0376727664d9422041dfbd18 (patch) | |
tree | d1a94be893451a4e182f3e75e9afb01749172bb4 /store.py | |
parent | f9226903cb53a9b303a26de562e321159349f8df (diff) | |
parent | dc1ca9a306f09886c6c57f8d426c59a9d084b2b3 (diff) | |
download | Trader-34eb08f759a440af0376727664d9422041dfbd18.tar.gz Trader-34eb08f759a440af0376727664d9422041dfbd18.tar.zst Trader-34eb08f759a440af0376727664d9422041dfbd18.zip |
Merge branch 'immae/parallelize' into dev
Fixes https://git.immae.eu/mantisbt/view.php?id=51
Diffstat (limited to 'store.py')
-rw-r--r-- | store.py | 175 |
1 files changed, 172 insertions, 3 deletions
@@ -1,10 +1,14 @@ | |||
1 | import time | ||
2 | import requests | ||
1 | import portfolio | 3 | import portfolio |
2 | import simplejson as json | 4 | import simplejson as json |
3 | from decimal import Decimal as D, ROUND_DOWN | 5 | from decimal import Decimal as D, ROUND_DOWN |
4 | from datetime import date, datetime | 6 | from datetime import date, datetime, timedelta |
5 | import inspect | 7 | import inspect |
8 | from json import JSONDecodeError | ||
9 | from simplejson.errors import JSONDecodeError as SimpleJSONDecodeError | ||
6 | 10 | ||
7 | __all__ = ["BalanceStore", "ReportStore", "TradeStore"] | 11 | __all__ = ["Portfolio", "BalanceStore", "ReportStore", "TradeStore"] |
8 | 12 | ||
9 | class ReportStore: | 13 | class ReportStore: |
10 | def __init__(self, market, verbose_print=True): | 14 | def __init__(self, market, verbose_print=True): |
@@ -13,6 +17,10 @@ class ReportStore: | |||
13 | 17 | ||
14 | self.logs = [] | 18 | self.logs = [] |
15 | 19 | ||
20 | def merge(self, other_report): | ||
21 | self.logs += other_report.logs | ||
22 | self.logs.sort(key=lambda x: x["date"]) | ||
23 | |||
16 | def print_log(self, message): | 24 | def print_log(self, message): |
17 | message = str(message) | 25 | message = str(message) |
18 | if self.verbose_print: | 26 | if self.verbose_print: |
@@ -213,7 +221,7 @@ class BalanceStore: | |||
213 | 221 | ||
214 | def dispatch_assets(self, amount, liquidity="medium", repartition=None): | 222 | def dispatch_assets(self, amount, liquidity="medium", repartition=None): |
215 | if repartition is None: | 223 | if repartition is None: |
216 | repartition = portfolio.Portfolio.repartition(self.market, liquidity=liquidity) | 224 | repartition = Portfolio.repartition(liquidity=liquidity) |
217 | sum_ratio = sum([v[0] for k, v in repartition.items()]) | 225 | sum_ratio = sum([v[0] for k, v in repartition.items()]) |
218 | amounts = {} | 226 | amounts = {} |
219 | for currency, (ptt, trade_type) in repartition.items(): | 227 | for currency, (ptt, trade_type) in repartition.items(): |
@@ -301,4 +309,165 @@ class TradeStore: | |||
301 | for order in self.all_orders(state="open"): | 309 | for order in self.all_orders(state="open"): |
302 | order.get_status() | 310 | order.get_status() |
303 | 311 | ||
312 | class NoopLock: | ||
313 | def __enter__(self, *args): | ||
314 | pass | ||
315 | def __exit__(self, *args): | ||
316 | pass | ||
317 | |||
318 | class LockedVar: | ||
319 | def __init__(self, value): | ||
320 | self.lock = NoopLock() | ||
321 | self.val = value | ||
322 | |||
323 | def start_lock(self): | ||
324 | import threading | ||
325 | self.lock = threading.Lock() | ||
326 | |||
327 | def set(self, value): | ||
328 | with self.lock: | ||
329 | self.val = value | ||
330 | |||
331 | def get(self, key=None): | ||
332 | with self.lock: | ||
333 | if key is not None and isinstance(self.val, dict): | ||
334 | return self.val.get(key) | ||
335 | else: | ||
336 | return self.val | ||
337 | |||
338 | def __getattr__(self, key): | ||
339 | with self.lock: | ||
340 | return getattr(self.val, key) | ||
341 | |||
342 | class Portfolio: | ||
343 | URL = "https://cryptoportfolio.io/wp-content/uploads/portfolio/json/cryptoportfolio.json" | ||
344 | data = LockedVar(None) | ||
345 | liquidities = LockedVar({}) | ||
346 | last_date = LockedVar(None) | ||
347 | report = LockedVar(ReportStore(None)) | ||
348 | worker = None | ||
349 | worker_started = False | ||
350 | worker_notify = None | ||
351 | callback = None | ||
352 | |||
353 | @classmethod | ||
354 | def start_worker(cls, poll=30): | ||
355 | import threading | ||
356 | |||
357 | cls.worker = threading.Thread(name="portfolio", daemon=True, | ||
358 | target=cls.wait_for_notification, kwargs={"poll": poll}) | ||
359 | cls.worker_notify = threading.Event() | ||
360 | cls.callback = threading.Event() | ||
361 | |||
362 | cls.last_date.start_lock() | ||
363 | cls.liquidities.start_lock() | ||
364 | cls.report.start_lock() | ||
365 | |||
366 | cls.worker_started = True | ||
367 | cls.worker.start() | ||
368 | |||
369 | @classmethod | ||
370 | def is_worker_thread(cls): | ||
371 | if cls.worker is None: | ||
372 | return False | ||
373 | else: | ||
374 | import threading | ||
375 | return cls.worker == threading.current_thread() | ||
376 | |||
377 | @classmethod | ||
378 | def wait_for_notification(cls, poll=30): | ||
379 | if not cls.is_worker_thread(): | ||
380 | raise RuntimeError("This method needs to be ran with the worker") | ||
381 | while cls.worker_started: | ||
382 | cls.worker_notify.wait() | ||
383 | cls.worker_notify.clear() | ||
384 | cls.report.print_log("Fetching cryptoportfolio") | ||
385 | cls.get_cryptoportfolio(refetch=True) | ||
386 | cls.callback.set() | ||
387 | time.sleep(poll) | ||
388 | |||
389 | @classmethod | ||
390 | def notify_and_wait(cls): | ||
391 | cls.callback.clear() | ||
392 | cls.worker_notify.set() | ||
393 | cls.callback.wait() | ||
394 | |||
395 | @classmethod | ||
396 | def wait_for_recent(cls, delta=4, poll=30): | ||
397 | cls.get_cryptoportfolio() | ||
398 | while cls.last_date.get() is None or datetime.now() - cls.last_date.get() > timedelta(delta): | ||
399 | if cls.worker is None: | ||
400 | time.sleep(poll) | ||
401 | cls.report.print_log("Attempt to fetch up-to-date cryptoportfolio") | ||
402 | cls.get_cryptoportfolio(refetch=True) | ||
403 | |||
404 | @classmethod | ||
405 | def repartition(cls, liquidity="medium"): | ||
406 | cls.get_cryptoportfolio() | ||
407 | liquidities = cls.liquidities.get(liquidity) | ||
408 | return liquidities[cls.last_date.get()] | ||
409 | |||
410 | @classmethod | ||
411 | def get_cryptoportfolio(cls, refetch=False): | ||
412 | if cls.data.get() is not None and not refetch: | ||
413 | return | ||
414 | if cls.worker is not None and not cls.is_worker_thread(): | ||
415 | cls.notify_and_wait() | ||
416 | return | ||
417 | try: | ||
418 | r = requests.get(cls.URL) | ||
419 | cls.report.log_http_request(r.request.method, | ||
420 | r.request.url, r.request.body, r.request.headers, r) | ||
421 | except Exception as e: | ||
422 | cls.report.log_error("get_cryptoportfolio", exception=e) | ||
423 | return | ||
424 | try: | ||
425 | cls.data.set(r.json(parse_int=D, parse_float=D)) | ||
426 | cls.parse_cryptoportfolio() | ||
427 | except (JSONDecodeError, SimpleJSONDecodeError): | ||
428 | cls.data.set(None) | ||
429 | cls.last_date.set(None) | ||
430 | cls.liquidities.set({}) | ||
431 | |||
432 | @classmethod | ||
433 | def parse_cryptoportfolio(cls): | ||
434 | def filter_weights(weight_hash): | ||
435 | if weight_hash[1][0] == 0: | ||
436 | return False | ||
437 | if weight_hash[0] == "_row": | ||
438 | return False | ||
439 | return True | ||
440 | |||
441 | def clean_weights(i): | ||
442 | def clean_weights_(h): | ||
443 | if h[0].endswith("s"): | ||
444 | return [h[0][0:-1], (h[1][i], "short")] | ||
445 | else: | ||
446 | return [h[0], (h[1][i], "long")] | ||
447 | return clean_weights_ | ||
448 | |||
449 | def parse_weights(portfolio_hash): | ||
450 | if "weights" not in portfolio_hash: | ||
451 | return {} | ||
452 | weights_hash = portfolio_hash["weights"] | ||
453 | weights = {} | ||
454 | for i in range(len(weights_hash["_row"])): | ||
455 | date = datetime.strptime(weights_hash["_row"][i], "%Y-%m-%d") | ||
456 | weights[date] = dict(filter( | ||
457 | filter_weights, | ||
458 | map(clean_weights(i), weights_hash.items()))) | ||
459 | return weights | ||
460 | |||
461 | high_liquidity = parse_weights(cls.data.get("portfolio_1")) | ||
462 | medium_liquidity = parse_weights(cls.data.get("portfolio_2")) | ||
463 | |||
464 | cls.liquidities.set({ | ||
465 | "medium": medium_liquidity, | ||
466 | "high": high_liquidity, | ||
467 | }) | ||
468 | cls.last_date.set(max( | ||
469 | max(medium_liquidity.keys(), default=datetime(1, 1, 1)), | ||
470 | max(high_liquidity.keys(), default=datetime(1, 1, 1)) | ||
471 | )) | ||
472 | |||
304 | 473 | ||