Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self.entity_ids:
entity_ids = json.dumps(self.entity_ids)
else:
entity_ids = None
if self.exchanges:
exchanges = json.dumps(self.exchanges)
else:
exchanges = None
if self.codes:
codes = json.dumps(self.codes)
else:
codes = None
trader_domain = business.Trader(id=self.trader_name, timestamp=self.start_timestamp,
trader_name=self.trader_name,
entity_type=entity_ids, exchanges=exchanges, codes=codes,
start_timestamp=self.start_timestamp,
end_timestamp=self.end_timestamp, provider=self.provider,
level=self.level.value,
real_time=self.real_time, kdata_use_begin_time=self.kdata_use_begin_time)
self.session.add(trader_domain)
self.session.commit()
self.selectors_comparator = self.init_selectors_comparator()
self.trading_level_asc = list(set([IntervalLevel(selector.level) for selector in self.selectors]))
self.trading_level_asc.sort()
self.trading_level_desc = list(self.trading_level_asc)
self.trading_level_desc.reverse()
self.targets_slot: TargetsSlot = TargetsSlot()
self.session = get_db_session('zvt', 'business')
trader = get_trader(session=self.session, trader_name=self.trader_name, return_type='domain', limit=1)
if trader:
self.logger.warning("trader:{} has run before,old result would be deleted".format(self.trader_name))
self.session.query(business.Trader).filter(business.Trader.trader_name == self.trader_name).delete()
self.session.commit()
self.on_start()
import dash_core_components as dcc
import dash_html_components as html
from dash.dependencies import Input, Output
from zvt.api.business import get_trader
from zvdata.app import app
from zvt.charts.dcc_components import get_trader_detail_figures
from zvt.domain import business
from zvt.reader.business_reader import AccountReader, OrderReader
account_readers = []
order_readers = []
# init the data
trader_domains: List[business.Trader] = get_trader(return_type='domain')
trader_names: List[str] = [item.trader_name for item in trader_domains]
def load_traders():
global trader_domains
global trader_names
trader_domains = get_trader(return_type='domain')
for trader in trader_domains:
account_readers.append(AccountReader(trader_names=[trader.trader_name], level=trader.level))
order_readers.append(OrderReader(trader_names=[trader.trader_name]))
trader_names = [item.trader_name for item in trader_domains]
import dash_core_components as dcc
import dash_html_components as html
from dash.dependencies import Input, Output
from zvt.api.business import get_trader
from zvt.app import app
from zvt.drawer.dcc_components import get_trader_detail_figures
from zvt.domain import business
from zvt.reader.business_reader import AccountReader, OrderReader
account_readers = []
order_readers = []
# init the data
trader_domains: List[business.Trader] = get_trader(return_type='domain')
trader_names: List[str] = [item.trader_name for item in trader_domains]
def load_traders():
global trader_domains
global trader_names
trader_domains = get_trader(return_type='domain')
for trader in trader_domains:
account_readers.append(AccountReader(trader_names=[trader.trader_name], level=trader.level))
order_readers.append(OrderReader(trader_names=[trader.trader_name]))
trader_names = [item.trader_name for item in trader_domains]
def get_trader(trader_name=None, return_type='df', start_timestamp=None, end_timestamp=None,
filters=None, session=None, order=None, limit=None) -> List[business.Trader]:
if trader_name:
if filters:
filters = filters + [business.Trader.trader_name == trader_name]
else:
filters = [business.Trader.trader_name == trader_name]
return get_data(data_schema=business.Trader, entity_id=None, codes=None, level=None, provider='zvt',
columns=None, return_type=return_type, start_timestamp=start_timestamp,
end_timestamp=end_timestamp, filters=filters, session=session, order=order, limit=limit)
def get_trader(trader_name=None, return_type='df', start_timestamp=None, end_timestamp=None,
filters=None, session=None, order=None, limit=None) -> List[business.Trader]:
if trader_name:
if filters:
filters = filters + [business.Trader.trader_name == trader_name]
else:
filters = [business.Trader.trader_name == trader_name]
return get_data(data_schema=business.Trader, entity_id=None, codes=None, level=None, provider='zvt',
columns=None, return_type=return_type, start_timestamp=start_timestamp,
end_timestamp=end_timestamp, filters=filters, session=session, order=order, limit=limit)
def get_trader(trader_name=None, return_type='df', start_timestamp=None, end_timestamp=None,
filters=None, session=None, order=None, limit=None) -> List[business.Trader]:
if trader_name:
if filters:
filters = filters + [business.Trader.trader_name == trader_name]
else:
filters = [business.Trader.trader_name == trader_name]
return get_data(data_schema=business.Trader, entity_id=None, codes=None, level=None, provider='zvt',
columns=None, return_type=return_type, start_timestamp=start_timestamp,
end_timestamp=end_timestamp, filters=filters, session=session, order=order, limit=limit)