同步汇率

basic/dao/robo_exrate.py

-from py_jftech import read, where, format_date
+from py_jftech import read, where, format_date, mapper_columns, write
 
 __COLUMNS__ = {
     're_id': 'id',
@@ -8,6 +8,13 @@ __COLUMNS__ = {
 }
 
 
+@write
+def batch_insert(datas):
+    datas = [mapper_columns(x, __COLUMNS__, ignore_none=False) for x in datas]
+    values = ','.join([f'''({','.join([(f"'{x[j]}'" if j in x and x[j] is not None else 'null') for j in __COLUMNS__.keys() if j != 're_id'])})''' for x in datas])
+    return f'''insert into robo_exrate({','.join([x for x in __COLUMNS__.keys() if x != 're_id'])}) values {values}'''
+
+
 @read
 def get_exrates(ticker=None, min_date=None, max_date=None):
     sqls = []
@@ -21,3 +28,12 @@ def get_exrates(ticker=None, min_date=None, max_date=None):
 @read(one=True)
 def get_exrate(ticker, date):
     return f'''select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_exrate {where(re_ticker=ticker, re_date=date)}'''
+
+
+@read(one=True)
+def get_last_one(ticker, max_date=None):
+    sql = f"re_date <= '{format_date(max_date)}'" if max_date else None
+    return f'''
+    select {','.join([f"{x[0]} as {x[1]}" for x in __COLUMNS__.items()])} from robo_exrate
+    {where(sql, re_ticker=ticker)} order by re_date desc limit 1
+    '''

basic/sync.py

@@ -8,7 +8,7 @@ import requests
 from py_jftech import format_date, is_workday, component, autowired, get_config, filter_weekend, next_workday
 
 from api import DatumType, DataSync, Datum
-from basic.dao import robo_index_datas as rid, robo_eco_datas as red, robo_fund_navs as rfn
+from basic.dao import robo_index_datas as rid, robo_eco_datas as red, robo_fund_navs as rfn, robo_exrate as re
 
 logger = logging.getLogger(__name__)
 
@@ -164,3 +164,48 @@ class FundNavSync(JDCDataSync):
         } for x in datas if is_workday(dt.fromtimestamp(x['date'] / 1000))]
         if save_navs:
             rfn.batch_insert(save_navs)
+
+
+@component(bean_name='exrate-sync')
+class ExrateSync(DataSync):
+
+    def __init__(self):
+        self._config = get_config(__name__)
+
+    @property
+    def start_date(self):
+        return filter_weekend(self._config['start-date'])
+
+    @property
+    def exrate_tickers(self):
+        navs_config = get_config('basic.navs')
+        return [x['ticker'] for x in navs_config['exrate']] if 'exrate' in navs_config else []
+
+    def build_url(self, ticker, start_date, page=0):
+        return f'http://jdcprod.thiztech.com/api/datas/exrate-value?page={page}&size=200&sourceCode={quote(ticker)}&sourceType=BLOOMBERG&startDate={format_date(start_date)}'
+
+    def do_sync(self, max_date=dt.today()):
+        logger.info(f'start sync datas for type[EXRATE]')
+        for ticker in self.exrate_tickers:
+            last_one = re.get_last_one(ticker=ticker, max_date=max_date)
+            start_date = next_workday(last_one['date']) if last_one else self.start_date
+            page = 0
+            while True:
+                url = self.build_url(ticker=ticker, start_date=start_date, page=page)
+                response = requests.get(url).json()
+                if not response['success']:
+                    raise Exception(f'''request indictor failed: {response['status']}''')
+                try:
+                    save_dates = [{
+                        'ticker': ticker,
+                        'date': dt.fromtimestamp(x['date'] / 1000),
+                        'close': x['close'],
+                    } for x in response['body']['content']]
+                    re.batch_insert(save_dates)
+                except Exception as e:
+                    logger.exception(f'url[{url}] store data failed')
+                    raise e
+                if response['body']['last']:
+                    break
+                else:
+                    page += 1

basic/test_case.py

@@ -33,6 +33,10 @@ class BasicTest(unittest.TestCase):
     def test_navs_sync(self, sync: DataSync = None):
         sync.do_sync()
 
+    @autowired(names={'sync': 'exrate-sync'})
+    def test_navs_sync(self, sync: DataSync = None):
+        sync.do_sync()
+
     @autowired(names={'report': 'navs-report'})
     def test_export_navs(self, report: RoboReportor = None):
         result = report.load_report()