Skip to content

Commit

Permalink
更新上期所行情接口
Browse files Browse the repository at this point in the history
更新上期所行情接口
  • Loading branch information
timercrack committed May 13, 2024
1 parent 38726c5 commit fe2e12c
Show file tree
Hide file tree
Showing 4 changed files with 26 additions and 41 deletions.
53 changes: 19 additions & 34 deletions test/test.py
Original file line number Diff line number Diff line change
@@ -1,44 +1,29 @@
import aioredis
import sys
import os
import django
if sys.platform == 'darwin':
sys.path.append('/Users/jeffchen/Documents/gitdir/dashboard')
elif sys.platform == 'win32':
sys.path.append(r'E:\github\dashboard')
else:
sys.path.append('/root/gitee/dashboard')
os.environ["DJANGO_SETTINGS_MODULE"] = "dashboard.settings"
os.environ["DJANGO_ALLOW_ASYNC_UNSAFE"] = "true"
django.setup()
import asyncio
import datetime
import pytz
from trader.utils import update_from_czce


async def reader(message):
print(f"in reader: {message}, type={type(message)}")


async def monitor1(pb: aioredis.client.PubSub):
async for msg in pb.listen():
print(f"in monitor1: {msg}, type={type(msg)}")
asyncio.create_task(reader(msg))
if msg['type'] == 'punsubscribe':
print('quit monitor1')
break


async def main():
redis = await aioredis.from_url("redis://192.168.123.142", decode_responses=True)
pubsub1 = redis.pubsub()
pubsub2 = redis.pubsub()
await pubsub1.psubscribe('channel:*')
await pubsub2.psubscribe('channel*')
await redis.publish("channel:1", "1")
await redis.publish("channel:2", "6")
await redis.publish("channel:3", "3")
await pubsub1.punsubscribe()
await pubsub2.punsubscribe()
loop = asyncio.get_running_loop()
asyncio.run_coroutine_threadsafe(monitor1(pubsub1), loop)
asyncio.run_coroutine_threadsafe(monitor1(pubsub2), loop)
# await asyncio.gather(monitor1(pubsub1), monitor1(pubsub2))
# await pubsub2.run()
print('main done!')

if __name__ == "__main__":
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.create_task(main())
loop.run_forever()
day = datetime.datetime.now().replace(tzinfo=pytz.FixedOffset(480))
day = day - datetime.timedelta(days=1)
loop.run_until_complete(update_from_czce(day))
print("DONE!")
except KeyboardInterrupt:
pass

6 changes: 3 additions & 3 deletions test/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
if sys.platform == 'darwin':
sys.path.append('/Users/jeffchen/Documents/gitdir/dashboard')
elif sys.platform == 'win32':
sys.path.append(r'D:\GitHub\dashboard')
sys.path.append(r'E:\GitHub\dashboard')
else:
sys.path.append('/root/dashboard')
os.environ["DJANGO_SETTINGS_MODULE"] = "dashboard.settings"
Expand All @@ -45,7 +45,7 @@ def tearDown(self) -> None:
async def test_get_shfe_data(self):
self.assertTrue(await update_from_shfe(self.trading_day))

@asynctest.skipIf(True, 'no need')
@asynctest.skipIf(False, 'no need')
async def test_get_dce_data(self):
self.assertTrue(await update_from_dce(self.trading_day))

Expand Down Expand Up @@ -74,7 +74,7 @@ async def test_get_all(self):
result = await asyncio.gather(*tasks, return_exceptions=True)
self.assertEqual(result, [True, True, True, True, True])

@asynctest.skipIf(False, 'no need')
@asynctest.skipIf(True, 'no need')
async def test_load_from_kt(self):
self.assertTrue(load_kt_data(r'D:\test'))

Expand Down
2 changes: 1 addition & 1 deletion trader/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
if sys.platform == 'darwin':
sys.path.append('/Users/jeffchen/Documents/gitdir/dashboard')
elif sys.platform == 'win32':
sys.path.append(r'D:\github\dashboard')
sys.path.append(r'E:\github\dashboard')
else:
sys.path.append('/root/gitee/dashboard')
os.environ["DJANGO_SETTINGS_MODULE"] = "dashboard.settings"
Expand Down
6 changes: 3 additions & 3 deletions trader/utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ async def update_from_dce(day: datetime.datetime) -> bool:
'year': day.year, 'month': day.month-1, 'day': day.day}) as response:
rst = await response.text()
max_conn_dce.release()
for lines in rst.split('\r\n')[1:-3]:
for lines in rst.split('\r\n')[3:-3]:
if '小计' in lines or '品种' in lines:
continue
inst_data_raw = [x.strip() for x in lines.split('\t')]
Expand Down Expand Up @@ -297,7 +297,7 @@ async def update_from_cffex(day: datetime.datetime) -> bool:
try:
async with aiohttp.ClientSession() as session:
await max_conn_cffex.acquire()
async with session.get(f"http://{cffex_ip}/fzjy/mrhq/{day.strftime('%Y%m/%d')}/index.xml") as response:
async with session.get(f"http://{cffex_ip}/sj/hqsj/rtj/{day.strftime('%Y%m/%d')}/index.xml?id=7") as response:
rst = await response.text()
max_conn_cffex.release()
tree = ET.fromstring(rst)
Expand Down Expand Up @@ -693,7 +693,7 @@ async def get_contracts_argument(day: datetime.datetime = None) -> bool:
async with aiohttp.ClientSession() as session:
# 上期所
async with session.get(
f'http://{shfe_ip}/data/instrument/ContractDailyTradeArgument{day_str}.dat') as response:
f'http://{shfe_ip}/data/busiparamdata/future/ContractDailyTradeArgument{day_str}.dat') as response:
rst = await response.read()
rst_json = json.loads(rst)
for inst_data in rst_json['ContractDailyTradeArgument']:
Expand Down

0 comments on commit fe2e12c

Please sign in to comment.