Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _update_dry(self) -> None:
"""
Update from database in dry-run mode
- Apply apply profits of closed trades on top of stake amount
- Subtract currently tied up stake_amount in open trades
- update balances for currencies currently in trades
"""
# Recreate _wallets to reset closed trade balances
_wallets = {}
closed_trades = Trade.get_trades(Trade.is_open.is_(False)).all()
open_trades = Trade.get_trades(Trade.is_open.is_(True)).all()
tot_profit = sum([trade.calc_profit() for trade in closed_trades])
tot_in_trades = sum([trade.stake_amount for trade in open_trades])
current_stake = self.start_cap + tot_profit - tot_in_trades
_wallets[self._config['stake_currency']] = Wallet(
self._config['stake_currency'],
current_stake,
0,
current_stake
)
for trade in open_trades:
curr = trade.pair.split('/')[0]
_wallets[curr] = Wallet(
curr,
today = datetime.utcnow().date()
profit_days = {}
try:
timescale = int(update.message.text.replace('/daily', '').strip())
except (TypeError, ValueError):
timescale = 7
if not (isinstance(timescale, int) and timescale > 0):
send_msg('*Daily [n]:* `must be an integer greater than 0`', bot=bot)
return
for day in range(0, timescale):
profitday = today - timedelta(days=day)
trades = Trade.query \
.filter(Trade.is_open.is_(False)) \
.filter(Trade.close_date >= profitday)\
.filter(Trade.close_date < (profitday + timedelta(days=1)))\
.order_by(Trade.close_date)\
.all()
curdayprofit = sum(trade.calc_profit() for trade in trades)
profit_days[profitday] = format(curdayprofit, '.8f')
stats = [
[
key,
'{value:.8f} {symbol}'.format(value=float(value), symbol=_CONF['stake_currency']),
'{value:.3f} {symbol}'.format(
value=_FIAT_CONVERT.convert_amount(
value,
_CONF['stake_currency'],
_CONF['fiat_display_currency']
def _rpc_daily_profit(
self, timescale: int,
stake_currency: str, fiat_display_currency: str) -> List[List[Any]]:
today = datetime.utcnow().date()
profit_days: Dict[date, Dict] = {}
if not (isinstance(timescale, int) and timescale > 0):
raise RPCException('timescale must be an integer greater than 0')
for day in range(0, timescale):
profitday = today - timedelta(days=day)
trades = Trade.get_trades(trade_filter=[
Trade.is_open.is_(False),
Trade.close_date >= profitday,
Trade.close_date < (profitday + timedelta(days=1))
]).order_by(Trade.close_date).all()
curdayprofit = sum(trade.calc_profit() for trade in trades)
profit_days[profitday] = {
'amount': f'{curdayprofit:.8f}',
'trades': len(trades)
}
return [
[
key,
'{value:.8f} {symbol}'.format(
value=float(value['amount']),
symbol=stake_currency
),
def _status_table(bot: Bot, update: Update) -> None:
"""
Handler for /status table.
Returns the current TradeThread status in table format
:param bot: telegram bot
:param update: message update
:return: None
"""
# Fetch open trade
trades = Trade.query.filter(Trade.is_open.is_(True)).all()
if get_state() != State.RUNNING:
send_msg('*Status:* `trader is not running`', bot=bot)
elif not trades:
send_msg('*Status:* `no active order`', bot=bot)
else:
trades_list = []
for trade in trades:
# calculate profit and send message to user
current_rate = exchange.get_ticker(trade.pair)['bid']
trades_list.append([
trade.id,
trade.pair,
shorten_date(arrow.get(trade.open_date).humanize(only_distance=True)),
'{:.2f}%'.format(100 * trade.calc_profit_percent(current_rate))
])
execute_sell(trade, current_rate)
# ---- EOF def _exec_forcesell ----
if get_state() != State.RUNNING:
return (True, '`trader is not running`')
if trade_id == 'all':
# Execute sell for all open orders
for trade in Trade.query.filter(Trade.is_open.is_(True)).all():
_exec_forcesell(trade)
return (False, '')
# Query for trade
trade = Trade.query.filter(sql.and_(
Trade.id == trade_id,
Trade.is_open.is_(True)
)).first()
if not trade:
logger.warning('forcesell: Invalid argument received')
return (True, 'Invalid argument.')
_exec_forcesell(trade)
return (False, '')
def get_open_trades() -> List[Any]:
"""
Query trades from persistence layer
"""
return Trade.get_trades(Trade.is_open.is_(True)).all()
def _performance(bot: Bot, update: Update) -> None:
"""
Handler for /performance.
Shows a performance statistic from finished trades
:param bot: telegram bot
:param update: message update
:return: None
"""
if get_state() != State.RUNNING:
send_msg('`trader is not running`', bot=bot)
return
pair_rates = Trade.session.query(Trade.pair, func.sum(Trade.close_profit).label('profit_sum'),
func.count(Trade.pair).label('count')) \
.filter(Trade.is_open.is_(False)) \
.group_by(Trade.pair) \
.order_by(text('profit_sum DESC')) \
.all()
stats = '\n'.join('{index}.\t<code>{pair}\t{profit:.2f}% ({count})</code>'.format(
index=i + 1,
pair=pair,
#url=exchange.get_pair_detail_url(pair),
profit=round(rate * 100, 2),
count=count
) for i, (pair, rate, count) in enumerate(pair_rates))
message = '<b>Performance:</b>\n{}'.format(stats)
logger.debug(message)
send_msg(message, parse_mode=ParseMode.HTML)
def rpc_status_table():
trades = Trade.query.filter(Trade.is_open.is_(True)).all()
if get_state() != State.RUNNING:
return (True, '*Status:* `trader is not running`')
elif not trades:
return (True, '*Status:* `no active order`')
else:
trades_list = []
for trade in trades:
# calculate profit and send message to user
current_rate = exchange.get_ticker(trade.pair, False)['bid']
trades_list.append([
trade.id,
trade.pair,
shorten_date(arrow.get(trade.open_date).humanize(only_distance=True)),
'{:.2f}%'.format(100 * trade.calc_profit_percent(current_rate))
])
"""
if get_state() != State.RUNNING:
send_msg('`trader is not running`', bot=bot)
return
trade_id = update.message.text.replace('/forcesell', '').strip()
if trade_id == 'all':
# Execute sell for all open orders
for trade in Trade.query.filter(Trade.is_open.is_(True)).all():
_exec_forcesell(trade)
return
# Query for trade
trade = Trade.query.filter(and_(
Trade.id == trade_id,
Trade.is_open.is_(True)
)).first()
if not trade:
send_msg('Invalid argument. See `/help` to view usage')
logger.warning('/forcesell: Invalid argument received')
return
_exec_forcesell(trade)