Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _timeout(i):
from app import app
if app.config.get('TESTING'):
return 0
return 2 ** i
def wait(wait_range=15):
wait_time = random.choice(range(wait_range))
logger.info('Wait %s seconds.', wait_time)
time.sleep(wait_time)
requests_retry = retry(RETRY_COUNT, errors=requests.RequestException,
timeout=_timeout)
class IntercomError(Exception):
""" Base exception for IntercomClient.
"""
class IntercomValidationError(IntercomError):
""" Incorrect value for one of the request's parameter.
"""
class IntercomClient:
""" Client for making requests to the Intercom's API.
Ref: https://developers.intercom.io/docs
@retry(10, errors=json.decoder.JSONDecodeError, timeout=lambda a: 2 ** a)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def cancel_all_OrdersPending(self, ordertype, long_short=None):
"""
撤销全部挂单
ordertype: LIMIT,STOP,MARKET_IF_TOUCHED,
buy_sell: LONG, SHORT
"""
rv = self.get_OrdersPending()
rv = [dict(id=i.get('id'), units=i.get('units'))
for i in rv['orders'] if i['type'] in ordertype and i.get('units')]
if long_short is 'LONG':
idsToCancel = [order.get('id') for order in rv
if float(order['units']) > 0]
elif long_short is 'SHORT':
@retry(10, errors=json.decoder.JSONDecodeError, timeout=lambda a: 2 ** a)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def OrderCreate_pending(self, instrument, units, price, takeprofit=None,
stoploss=None, trailingstop=None,
requesttype='MarketIfTouchedOrder'):
"""
建立挂单
requesttype:
LimitOrder, StopOrder, MarketIfTouchedOrder,
"""
d = dict(instrument=instrument, units=units, price=price)
if takeprofit:
d['takeProfitOnFill'] = TakeProfitDetails(price=takeprofit).data
if stoploss:
d['stopLossOnFill'] = StopLossDetails(price=stoploss).data
if trailingstop:
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_tradeslist(self, instrument):
r = trades.TradesList(accountID=self.accountID, params={'instrument': instrument})
return self.client.request(r)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_AccountSummary(self):
r = accounts.AccountSummary(accountID=self.accountID)
return self.client.request(r)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_OrdersPending(self):
r = orders.OrdersPending(accountID=self.accountID)
return self.client.request(r)
@retry(10, errors=json.decoder.JSONDecodeError, timeout=lambda a: 2 ** a)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_AccountDetails(self):
r = accounts.AccountDetails(accountID=self.accountID)
return self.client.request(r)
@retry(10, errors=json.decoder.JSONDecodeError, timeout=lambda a: 2 ** a)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_OrderList(self, instrument):
"""
可以获得特定instrument的 Pending Order
"""
r = orders.OrderList(accountID=self.accountID,
params={"instrument": instrument})
return self.client.request(r)
@retry(50, errors=requests.HTTPError, timeout=30)
def _ensure_refmrna(specie):
specie_refmrna = os.path.join(settings.BASE_DIR, '_files/%s.fa' % specie)
if not os.path.isfile(specie_refmrna):
cprint('Downloading %s refMrna...' % specie, 'blue')
http_to_file(REFMRNA_URLS[specie], specie_refmrna + '.gz')
subprocess.check_call(['gzip', '-d', specie_refmrna + '.gz'],
stdout=sys.stdout, stderr=sys.stderr)
return specie_refmrna
@retry(10, errors=json.decoder.JSONDecodeError, timeout=lambda a: 2 ** a)
@retry(10, errors=oandapyV20.exceptions.V20Error, timeout=lambda a: 2 ** a)
@retry(10, errors=requests.exceptions.ProxyError, timeout=lambda a: 2 ** a)
def get_all_open_trades(self):
r = trades.OpenTrades(accountID=self.accountID)
return self.client.request(r)