diff --git a/QUANTAXIS_Test/QAARP_Test/QAAccount_Test.py b/QUANTAXIS_Test/QAARP_Test/QAAccount_Test.py
index ca90ed474..6e1c76cb5 100644
--- a/QUANTAXIS_Test/QAARP_Test/QAAccount_Test.py
+++ b/QUANTAXIS_Test/QAARP_Test/QAAccount_Test.py
@@ -1,33 +1,33 @@
-import unittest;
+import random as rnd
+import unittest
-from QUANTAXIS.QAUtil import (QADate)
-from QUANTAXIS.QAUtil import (QARandom)
import QUANTAXIS as QA
-import random as rnd
+from QUANTAXIS.QAUtil import QADate, QARandom
-class Test_QAAccount(unittest.TestCase):
+class Test_QAAccount(unittest.TestCase):
def test_BuyAndSell(self):
# 测试买卖事件
- #测试流程，
- #随机选中一组股票
- #随机选中几个交易日 (只测试当前日期）
- #随机卖出
- #随机买入
- #发送订单
- #从队列中取回订单
- #比较
+ # 测试流程，
+ # 随机选中一组股票
+ # 随机选中几个交易日 (只测试当前日期）
+ # 随机卖出
+ # 随机买入
+ # 发送订单
+ # 从队列中取回订单
+ # 比较
codeCount = 10
codeList = QARandom.QA_util_random_with_zh_stock_code(codeCount)
- #print(codeList)
+ # print(codeList)
# { code: QA_DataStruct_Stock_day }
# 如果代码不存在 则 code：None
codeForDate = {}
- for codeIndex in range(0,codeCount):
- aStockDataStructDay = QA.QA_fetch_stock_day_adv(codeList[codeIndex])
+ for codeIndex in range(0, codeCount):
+ aStockDataStructDay = QA.QA_fetch_stock_day_adv(
+ codeList[codeIndex])
codeForDate[codeList[codeIndex]] = aStockDataStructDay
- #print(codeForDate)
+ # print(codeForDate)
str = QADate.QA_util_today_str()
timestamp = QADate.QA_util_to_datetime(str)
@@ -37,18 +37,18 @@ def test_BuyAndSell(self):
orderList = []
for a_stock_code in codeForDate.keys():
anOrder = account.send_order(code=a_stock_code,
- amount=100,
- time = timestamp,
- towards= 1,
- price= 8.8,
- order_model= QA.ORDER_MODEL.LIMIT,
- amount_model=QA.AMOUNT_MODEL.BY_AMOUNT)
+ amount=100,
+ time=timestamp,
+ towards=1,
+ price=8.8,
+ order_model=QA.ORDER_MODEL.LIMIT,
+ amount_model=QA.AMOUNT_MODEL.BY_AMOUNT)
orderList.append(anOrder)
orderQueue = account.get_orders()
print(orderQueue)
- self.assertEqual(len(orderQueue.queue_df) , codeCount)
+ self.assertEqual(len(orderQueue.queue_df), codeCount)
orderList2 = []
print(orderQueue.queue_df)
@@ -56,7 +56,7 @@ def test_BuyAndSell(self):
anOrder2 = orderQueue.query_order(orderId)
orderList2.append(anOrder2)
- self.assertEqual( len(orderList),len(orderList2) )
+ self.assertEqual(len(orderList), len(orderList2))
orderCount = len(orderList)
for i in range(orderCount):
@@ -64,7 +64,7 @@ def test_BuyAndSell(self):
order_01 = orderList[i]
order_02 = orderList2[i]
- #总是不正确
+ # 总是不正确
#b = order_01 is order_02
#b = order_01 == order_02
#b = (order_01.__dict__ == order_02.__dict__)
@@ -78,53 +78,54 @@ def test_BuyAndSell(self):
print(v1)
print(v2)
self.fail("订单数据不正确")
- #todo 继续研究为何不正确
+ # todo 继续研究为何不正确
#self.assertEqual(order_01, order_02)
pass
def n0_test_QAAccount_class(self):
- #测试流程 获取 美康生物 300439 的走势 从 2017年10月01日开始 到 2018年 4月30日
+ # 测试流程 获取 美康生物 300439 的走势 从 2017年10月01日开始 到 2018年 4月30日
#
test_stock_code = '300439'
- stock_price_list = QA.QA_fetch_stock_day(code = test_stock_code, start = '2017-10-01', end = '2018-04-30')
+ stock_price_list = QA.QA_fetch_stock_day(
+ code=test_stock_code, start='2017-10-01', end='2018-04-30')
#buy_list = []
print(stock_price_list)
print("---------开始测试买入------------")
#price_list_size = stock_price_list.size
- #生成随机数种子
+ # 生成随机数种子
rnd.seed(QA.QA_util_time_now().timestamp())
Account = QA.QA_Account()
B = QA.QA_BacktestBroker()
for aday_stock_price in stock_price_list:
- stock_code = aday_stock_price[0]
- price_open = aday_stock_price[1]
- price_high = aday_stock_price[2]
- price_low = aday_stock_price[3]
+ stock_code = aday_stock_price[0]
+ price_open = aday_stock_price[1]
+ price_high = aday_stock_price[2]
+ price_low = aday_stock_price[3]
price_close = aday_stock_price[4]
stock_volume = aday_stock_price[5]
- stock_turn = aday_stock_price[6]
+ stock_turn = aday_stock_price[6]
stock_timestamp = aday_stock_price[7]
- #print(type(stock_timestamp))
+ # print(type(stock_timestamp))
- dt = QADate.QA_util_pands_timestamp_to_datetime(stock_timestamp);
- date_time_to_buy = QADate.QA_util_datetime_to_strdatetime(dt);
+ dt = QADate.QA_util_pands_timestamp_to_datetime(stock_timestamp)
+ date_time_to_buy = QADate.QA_util_datetime_to_strdatetime(dt)
dt = QADate.QA_util_to_datetime(date_time_to_buy)
if price_low != price_high:
- price_diff = (price_high - price_low);
+ price_diff = (price_high - price_low)
self.assertTrue(price_diff >= 0, "最高价一定是非负数")
rand_value = rnd.random()
buy_price = price_low + price_diff * rand_value
print("{} 申报买入的成交价格 {}".format(dt, buy_price))
- #每天随机在开盘价格和收盘价格直接买入
+ # 每天随机在开盘价格和收盘价格直接买入
Order = Account.send_order(code=test_stock_code,
price=buy_price,
amount=100,
@@ -134,21 +135,22 @@ def n0_test_QAAccount_class(self):
order_model=QA.ORDER_MODEL.LIMIT,
amount_model=QA.AMOUNT_MODEL.BY_AMOUNT
)
- print('ORDER的占用资金: {}'.format((Order.amount * Order.price) * (1 + Account.commission_coeff)))
- print('账户剩余资金 :{}'.format(Account.cash_available)) ## ??
+ print('ORDER的占用资金: {}'.format(
+ (Order.amount * Order.price) * (1 + Account.commission_coeff)))
+ print('账户剩余资金 :{}'.format(Account.cash_available)) # ??
cash_available = Account.cash_available
print(cash_available)
- rec_mes=B.receive_order(QA.QA_Event(order=Order))
+ rec_mes = B.receive_order(QA.QA_Event(order=Order))
print(rec_mes)
Account.receive_deal(rec_mes)
- knock_down_price = rec_mes['body']['order']['price'];
- bid_price = round(buy_price, 2);
+ knock_down_price = rec_mes['body']['order']['price']
+ bid_price = round(buy_price, 2)
print("{} 获取订单的价格 {}".format(dt, buy_price))
self.assertEqual(knock_down_price, bid_price)
print('账户的可用资金 {}'.format(Account.cash_available))
- print('-----------------------------------------------')
\ No newline at end of file
+ print('-----------------------------------------------')
diff --git a/QUANTAXIS_Test/QAARP_Test/QAUser_Test.py b/QUANTAXIS_Test/QAARP_Test/QAUser_Test.py
index d98386be4..9d129b5da 100644
--- a/QUANTAXIS_Test/QAARP_Test/QAUser_Test.py
+++ b/QUANTAXIS_Test/QAARP_Test/QAUser_Test.py
@@ -1,6 +1,7 @@
-import QUANTAXIS as QA
import unittest
+import QUANTAXIS as QA
+
class Test_QAUser(unittest.TestCase):
def testQAUser(self):
@@ -21,9 +22,8 @@ def testQAUser(self):
p2 = user.get_portfolio(portfolio1)
ac2 = user.get_portfolio(portfolio1).get_account(ac1)
- self.assertEqual(p2,p)
- self.assertEqual(ac1,ac2)
+ self.assertEqual(p2, p)
+ self.assertEqual(ac1, ac2)
except:
print("Error")
-
diff --git a/QUANTAXIS_Test/QAAnalysis_Test/QAAnalysis_dataframe_Test.py b/QUANTAXIS_Test/QAAnalysis_Test/QAAnalysis_dataframe_Test.py
index 5f498b7b8..d53a5251f 100644
--- a/QUANTAXIS_Test/QAAnalysis_Test/QAAnalysis_dataframe_Test.py
+++ b/QUANTAXIS_Test/QAAnalysis_Test/QAAnalysis_dataframe_Test.py
@@ -2,10 +2,12 @@
import QUANTAXIS as QA
+
class Test_QAAnalysis_dataframe(unittest.TestCase):
def test_QAAnalysis_stock(self):
- data = QA.QA_fetch_stock_day_adv('600066', '2013-12-01', '2017-10-01') # [可选to_qfq(),to_hfq()]
+ data = QA.QA_fetch_stock_day_adv(
+ '600066', '2013-12-01', '2017-10-01') # [可选to_qfq(),to_hfq()]
s = QA.QAAnalysis_stock(data)
# s 的属性是( < QAAnalysis_Stock > )
@@ -23,7 +25,7 @@ def test_QAAnalysis_stock(self):
s.max # price的最大值
s.min # price的最小值
s.mad # price的平均绝对偏差
- #s.mode # price的众数(没啥用)
+ # s.mode # price的众数(没啥用)
s.price_diff # price的一阶差分
s.variance # price的方差
s.pvariance # price的样本方差
@@ -35,4 +37,4 @@ def test_QAAnalysis_stock(self):
s.kurtosis # price的偏度 (3阶中心距)
s.pct_change # price的百分比变化序列
- s.add_func(QA.QA_indicator_CCI) # 指标计算, 和DataStruct用法一致
\ No newline at end of file
+ s.add_func(QA.QA_indicator_CCI) # 指标计算, 和DataStruct用法一致
diff --git a/QUANTAXIS_Test/QABacktest_Test/QABacktestSimple_Test.py b/QUANTAXIS_Test/QABacktest_Test/QABacktestSimple_Test.py
index 0f0bbdd14..7b2f3794d 100644
--- a/QUANTAXIS_Test/QABacktest_Test/QABacktestSimple_Test.py
+++ b/QUANTAXIS_Test/QABacktest_Test/QABacktestSimple_Test.py
@@ -1,15 +1,15 @@
import unittest
-import QUANTAXIS as QA
-
import numpy as np
import pandas as pd
+import QUANTAXIS as QA
+
class QABacktestSimple_Test(unittest.TestCase):
# define the MACD strategy
- def MACD_JCSC(self,dataframe, SHORT=12, LONG=26, M=9):
+ def MACD_JCSC(self, dataframe, SHORT=12, LONG=26, M=9):
"""
1.DIF向上突破DEA，买入信号参考。
2.DIF向下跌破DEA，卖出信号参考。
@@ -26,7 +26,7 @@ def MACD_JCSC(self,dataframe, SHORT=12, LONG=26, M=9):
{'DIFF': DIFF, 'DEA': DEA, 'MACD': MACD, 'CROSS_JC': CROSS_JC, 'CROSS_SC': CROSS_SC, 'ZERO': ZERO})
def setUp(self):
- #准备数据
+ # 准备数据
# create account
self.Account = QA.QA_Account()
@@ -44,8 +44,8 @@ def setUp(self):
self.ind = self.data.add_func(self.MACD_JCSC)
# ind.xs('000001',level=1)['2018-01'].plot()
- self.data_forbacktest = self.data.select_time('2018-01-01', '2018-05-20')
-
+ self.data_forbacktest = self.data.select_time(
+ '2018-01-01', '2018-05-20')
def tearDown(self):
@@ -54,7 +54,7 @@ def tearDown(self):
print(self.Account.daily_hold)
# create Risk analysis
- Risk = QA.QA_Risk( self.Account)
+ Risk = QA.QA_Risk(self.Account)
print(Risk.message)
print(Risk.assets)
Risk.plot_assets_curve()
@@ -67,7 +67,8 @@ def tearDown(self):
self.Account.save()
Risk.save()
- account_info = QA.QA_fetch_account({'account_cookie': 'user_admin_macd'})
+ account_info = QA.QA_fetch_account(
+ {'account_cookie': 'user_admin_macd'})
account = QA.QA_Account().from_message(account_info[0])
print(account)
@@ -93,7 +94,8 @@ def test_simpleQABacktest(self):
order = self.Account.send_order(
code=item.data.code[0],
time=item.data.date[0],
- amount=self.Account.sell_available.get(item.code[0], 0),
+ amount=self.Account.sell_available.get(
+ item.code[0], 0),
towards=QA.ORDER_DIRECTION.SELL,
price=0,
order_model=QA.ORDER_MODEL.MARKET,
@@ -102,4 +104,3 @@ def test_simpleQABacktest(self):
self.Account.receive_deal(self.Broker.receive_order(
QA.QA_Event(order=order, market_data=item)))
self.Account.settle()
-
diff --git a/QUANTAXIS_Test/QABacktest_Test/QABacktest_Test.py b/QUANTAXIS_Test/QABacktest_Test/QABacktest_Test.py
index 0c0429f70..84a19ee69 100644
--- a/QUANTAXIS_Test/QABacktest_Test/QABacktest_Test.py
+++ b/QUANTAXIS_Test/QABacktest_Test/QABacktest_Test.py
@@ -1,19 +1,20 @@
import unittest
-from QUANTAXIS.QAARP.QARisk import QA_Risk,QA_Performance
-from QUANTAXIS.QAARP.QAUser import QA_User
-from QUANTAXIS.QABacktest.QABacktest import QA_Backtest
-from QUANTAXIS.QAARP.QAStrategy import QA_Strategy
-from QUANTAXIS.QAUtil.QAParameter import (AMOUNT_MODEL, MARKET_TYPE, FREQUENCE, ORDER_DIRECTION, ORDER_MODEL,DATASOURCE, OUTPUT_FORMAT)
-
-from QUANTAXIS.QAUtil.QALogs import QA_util_log_info
-from QUANTAXIS.QAUtil.QADate import QA_util_datetime_to_strdate
-from QUANTAXIS import QUANTAXIS as QA
+import pandas as pd
import tushare as QATs
-from QUANTAXIS.QAFetch.QATdx import QA_fetch_get_stock_day
-from QUANTAXIS.QAIndicator import QA_indicator_MA
-import pandas as pd
+from QUANTAXIS import QUANTAXIS as QA
+from QUANTAXIS.QAARP.QARisk import QA_Performance, QA_Risk
+from QUANTAXIS.QAARP.QAStrategy import QA_Strategy
+from QUANTAXIS.QAARP.QAUser import QA_User
+from QUANTAXIS.QABacktest.QABacktest import QA_Backtest
+from QUANTAXIS.QAFetch.QATdx import QA_fetch_get_stock_day
+from QUANTAXIS.QAIndicator import QA_indicator_MA
+from QUANTAXIS.QAUtil.QADate import QA_util_datetime_to_strdate
+from QUANTAXIS.QAUtil.QALogs import QA_util_log_info
+from QUANTAXIS.QAUtil.QAParameter import (AMOUNT_MODEL, DATASOURCE, FREQUENCE,
+ MARKET_TYPE, ORDER_DIRECTION,
+ ORDER_MODEL, OUTPUT_FORMAT)
class MAStrategy(QA_Strategy):
@@ -27,23 +28,24 @@ def __init__(self):
# self.time_to_Market_300439 = QA.QA_util_date_int2str(self.time_to_Market_300439)
# self.time_to_day = QA_util_datetime_to_strdate(QA.QA_util_date_today())
# print(self.time_to_Market_300439)
- #print(self.time_to_day)
+ # print(self.time_to_day)
self.time_to_Market_300439 = '2015-04-22'
self.time_to_day = '2018-05-01'
- self.df_from_Tdx = QA_fetch_get_stock_day('300439', self.time_to_Market_300439, self.time_to_day, '01')
- #print(self.df_from_Tdx)
+ self.df_from_Tdx = QA_fetch_get_stock_day(
+ '300439', self.time_to_Market_300439, self.time_to_day, '01')
+ # print(self.df_from_Tdx)
self.ma05 = QA_indicator_MA(self.df_from_Tdx, 5)
self.ma10 = QA_indicator_MA(self.df_from_Tdx, 10)
self.ma15 = QA_indicator_MA(self.df_from_Tdx, 15)
self.ma20 = QA_indicator_MA(self.df_from_Tdx, 20)
- #print(self.df5)
+ # print(self.df5)
def on_bar(self, event):
- sellavailable=self.sell_available
- #try:
+ sellavailable = self.sell_available
+ # try:
#strDbg = QA.QA_util_random_with_topic("MAStrategy.on_bar call")
#print(">-----------------------on bar----------------------------->", strDbg)
for item in event.market_data.code:
@@ -51,57 +53,55 @@ def on_bar(self, event):
current_date = self.current_time.date()
#print("on bar 当前日期是:", current_date )
- today_T = pd.Timestamp(current_date);
+ today_T = pd.Timestamp(current_date)
vma05 = self.ma05.at[today_T, 'MA']
vma10 = self.ma10.at[today_T, 'MA']
vma15 = self.ma15.at[today_T, 'MA']
vma20 = self.ma20.at[today_T, 'MA']
-
if vma05 > vma10 and vma10 > vma15 and vma15 > vma20:
- #print("均线多头排列")
+ # print("均线多头排列")
if self.sell_available is not None and self.sell_available.get(item, 0) == 0:
- print("*>> MAStrategy!on_bar event.send_order 买入 buy %d"%(100))
+ print("*>> MAStrategy!on_bar event.send_order 买入 buy %d" % (100))
print(event.send_order)
print(type(event.send_order))
event.send_order(account_id=self.account_cookie,
- amount=100,
- amount_model=AMOUNT_MODEL.BY_AMOUNT,
- time=self.current_time,
- code=item,
- price=0,
- order_model=ORDER_MODEL.MARKET,
- towards=ORDER_DIRECTION.BUY,
- market_type=self.market_type,
- frequence=self.frequence,
- broker_name=self.broker)
-
+ amount=100,
+ amount_model=AMOUNT_MODEL.BY_AMOUNT,
+ time=self.current_time,
+ code=item,
+ price=0,
+ order_model=ORDER_MODEL.MARKET,
+ towards=ORDER_DIRECTION.BUY,
+ market_type=self.market_type,
+ frequence=self.frequence,
+ broker_name=self.broker)
elif vma05 < vma10 and vma10 < vma15 and vma15 < vma20:
- #print("均线空头排列")
+ # print("均线空头排列")
if self.sell_available is not None and self.sell_available.get(item, 0) > 0:
- print("*>> MAStrategy!on_bar event.send_order 卖出 buy %d"%100)
+ print("*>> MAStrategy!on_bar event.send_order 卖出 buy %d" % 100)
event.send_order(account_id=self.account_cookie,
- amount=self.sell_available[item],
- amount_model=AMOUNT_MODEL.BY_AMOUNT,
- time=self.current_time,
- code=item,
- price=0,
- order_model=ORDER_MODEL.MARKET,
- towards=ORDER_DIRECTION.SELL,
- market_type=self.market_type,
- frequence=self.frequence,
- broker_name=self.broker)
+ amount=self.sell_available[item],
+ amount_model=AMOUNT_MODEL.BY_AMOUNT,
+ time=self.current_time,
+ code=item,
+ price=0,
+ order_model=ORDER_MODEL.MARKET,
+ towards=ORDER_DIRECTION.SELL,
+ market_type=self.market_type,
+ frequence=self.frequence,
+ broker_name=self.broker)
else:
- #print("均线交叉中")
+ # print("均线交叉中")
pass
#print("<-----------------------on bar-----------------------------<",strDbg)
#current_date = self.current_time.date
- #except:
+ # except:
# pass
@@ -115,17 +115,18 @@ def __init__(self, market_type, frequence, start, end, code_list, commission_fee
def after_success(self):
QA_util_log_info(self.account.history_table)
- #check if the history_table is empty list
+ # check if the history_table is empty list
if len(self.account.history_table) == 0:
- #没有交易历史记录，直接返回
+ # 没有交易历史记录，直接返回
return
- risk = QA_Risk(self.account, benchmark_code='000300', benchmark_type=MARKET_TYPE.INDEX_CN)
+ risk = QA_Risk(self.account, benchmark_code='000300',
+ benchmark_type=MARKET_TYPE.INDEX_CN)
print(risk().T)
risk.plot_assets_curve()
risk.plot_dailyhold()
risk.plot_signal()
- performance=QA_Performance(self.account)
+ performance = QA_Performance(self.account)
performance.plot_pnlmoney(performance.pnl_fifo)
performance.plot_pnlratio(performance.pnl_fifo)
self.account.save()
@@ -134,21 +135,20 @@ def after_success(self):
class Test_QABacktest(unittest.TestCase):
-
def testBacktraceTest(self):
self.run_daybacktest()
pass
def run_daybacktest(self):
#import QUANTAXIS as QA
- #print(QA.QA_fetch_stock_block_adv().code[0:5])
+ # print(QA.QA_fetch_stock_block_adv().code[0:5])
# self.stock_basics = QATs.get_stock_basics()
# self.time_to_Market_300439 = self.stock_basics.loc['300439', 'timeToMarket']
# self.time_to_Market_300439 = QA.QA_util_date_int2str(self.time_to_Market_300439)
# self.time_to_day = QA_util_datetime_to_strdate(QA.QA_util_date_today())
# print(self.time_to_Market_300439)
- #print(self.time_to_day)
- #QA.QA_util_time_now()
+ # print(self.time_to_day)
+ # QA.QA_util_time_now()
self.time_to_Market_300439 = '2015-04-22'
self.time_to_day = '2018-05-01'
@@ -166,6 +166,5 @@ def run_daybacktest(self):
print("结束回测！")
-
if __name__ == '__main__':
- Test_QABacktest().testBacktraceTest()
\ No newline at end of file
+ Test_QABacktest().testBacktraceTest()
diff --git a/QUANTAXIS_Test/QAData_Test/quotation_base_test.py b/QUANTAXIS_Test/QAData_Test/quotation_base_test.py
index d655ed372..c72b2281d 100644
--- a/QUANTAXIS_Test/QAData_Test/quotation_base_test.py
+++ b/QUANTAXIS_Test/QAData_Test/quotation_base_test.py
@@ -1,31 +1,21 @@
-import unittest
import datetime
+import unittest
+import pandas as pd
+
+from QUANTAXIS import QUANTAXIS as QA
from QUANTAXIS.QAData import (QA_DataStruct_Index_day, QA_DataStruct_Index_min,
QA_DataStruct_Stock_block,
QA_DataStruct_Stock_day, QA_DataStruct_Stock_min,
QA_DataStruct_Stock_transaction)
-from QUANTAXIS.QAFetch.QAQuery import (QA_fetch_index_day,
- QA_fetch_index_min,
- QA_fetch_stock_day,
- QA_fetch_stock_full,
- QA_fetch_stock_min)
-
-
-from QUANTAXIS.QAFetch.QAQuery import (QA_fetch_index_day,
- QA_fetch_index_min,
- QA_fetch_stock_day,
- QA_fetch_stock_full,
+from QUANTAXIS.QAData.base_datastruct import _quotation_base
+from QUANTAXIS.QAFetch.QAQuery import (QA_fetch_index_day, QA_fetch_index_min,
+ QA_fetch_stock_day, QA_fetch_stock_full,
QA_fetch_stock_min)
-
from QUANTAXIS.QAFetch.QAQuery_Advance import QA_fetch_stock_day_adv
-
-import pandas as pd
-from QUANTAXIS.QAData.base_datastruct import _quotation_base
-from QUANTAXIS.QAUtil.QAParameter import FREQUENCE, MARKET_TYPE, DATASOURCE, OUTPUT_FORMAT, DATABASE_TABLE
-
-from QUANTAXIS import QUANTAXIS as QA
-
+from QUANTAXIS.QAUtil.QAParameter import (DATABASE_TABLE, DATASOURCE,
+ FREQUENCE, MARKET_TYPE,
+ OUTPUT_FORMAT)
class quotation_base_test(unittest.TestCase):
@@ -171,4 +161,11 @@ def test_00(self):
# 🛠todo 测试 loc
# 🛠todo 测试 iloc#
# 🛠todo 测试 iloc#
- # 🛠todo 测试 iloc
\ No newline at end of file
+ # 🛠todo 测试 iloc
+
+
+if __name__ == '__main__':
+
+ t = quotation_base_test()
+ t.setUp()
+ #t.testBacktestBollingerBands()
diff --git a/QUANTAXIS_Test/QAEngine_Test/QAEvent_Test.py b/QUANTAXIS_Test/QAEngine_Test/QAEvent_Test.py
index 29b55d67a..0164c487f 100644
--- a/QUANTAXIS_Test/QAEngine_Test/QAEvent_Test.py
+++ b/QUANTAXIS_Test/QAEngine_Test/QAEvent_Test.py
@@ -3,12 +3,14 @@
import QUANTAXIS as QA
+
"""
在这里 我们演示两种方法
1. 直接通过QA_Thread 创建一个事件线程做任务
2. 通过QA_Engine 来创建一个QA_Thread 来分派事件
"""
+
class job(QA.QA_Worker):
def __init__(self):
super().__init__()
@@ -31,13 +33,14 @@ def testThread(self):
engine.start() # engine 开启
engine.create_kernel('backtest') # engine创建一个叫 backtest的线程
- engine.start_kernel ('backtest') # engine 启动该线程
+ engine.start_kernel('backtest') # engine 启动该线程
# 创建一个类,继承QA_Worker
jobx = job() # 实例化这个类
# 创建一个event
- event = QA.QA_Event(event_type='selfdesign', message='ssss', callback=print)
+ event = QA.QA_Event(event_type='selfdesign',
+ message='ssss', callback=print)
# 创建一个标准task
task = QA.QA_Task(event=event, worker=jobx, engine='backtest')
@@ -51,4 +54,4 @@ def testThread(self):
engine.run_job(task) # 向engine推送任务
- pass
\ No newline at end of file
+ pass
diff --git a/QUANTAXIS_Test/QAFetch_Test/Fetcher_Test.py b/QUANTAXIS_Test/QAFetch_Test/Fetcher_Test.py
index 61829b789..9b514d2d7 100644
--- a/QUANTAXIS_Test/QAFetch_Test/Fetcher_Test.py
+++ b/QUANTAXIS_Test/QAFetch_Test/Fetcher_Test.py
@@ -9,39 +9,37 @@
import unittest
-from QUANTAXIS.QAFetch import QAWind as QAWind
-from QUANTAXIS.QAFetch import QATushare as QATushare
+from QUANTAXIS import QUANTAXIS as QA
+from QUANTAXIS import *
+from QUANTAXIS.QAFetch import QAEastMoney as QAEM
+from QUANTAXIS.QAFetch import QAQuery_Advance as QAQueryAdv
from QUANTAXIS.QAFetch import QATdx as QATdx
from QUANTAXIS.QAFetch import QAThs as QAThs
+from QUANTAXIS.QAFetch import QATushare as QATushare
+from QUANTAXIS.QAFetch import QAWind as QAWind
from QUANTAXIS.QAFetch import QAQuery
-from QUANTAXIS.QAFetch import QAQuery_Advance as QAQueryAdv
-from QUANTAXIS.QAFetch import QAEastMoney as QAEM
-from QUANTAXIS.QAUtil.QAParameter import FREQUENCE, MARKET_TYPE, DATASOURCE, OUTPUT_FORMAT, DATABASE_TABLE
+from QUANTAXIS.QAUtil.QAParameter import (DATABASE_TABLE, DATASOURCE,
+ FREQUENCE, MARKET_TYPE,
+ OUTPUT_FORMAT)
from QUANTAXIS.QAUtil.QASql import QA_util_sql_mongo_setting
-from QUANTAXIS import QUANTAXIS as QA
-
-
-from QUANTAXIS import *;
-
class Test_Fetcher(unittest.TestCase):
def test_QA_quotation(self):
#print("ok start test")
df_from_Tdx = QA_quotation('300439', '2018-04-01', '2018-04-10', frequence=FREQUENCE.DAY,
- market=MARKET_TYPE.STOCK_CN, source=DATASOURCE.TDX, output=OUTPUT_FORMAT.DATAFRAME)
+ market=MARKET_TYPE.STOCK_CN, source=DATASOURCE.TDX, output=OUTPUT_FORMAT.DATAFRAME)
- #print(type(df_from_Tdx))
+ # print(type(df_from_Tdx))
print((df_from_Tdx))
list_from_tushare = QA_quotation('300439', '2018-04-01', '2018-04-10', frequence=FREQUENCE.DAY,
- market=MARKET_TYPE.STOCK_CN, source=DATASOURCE.TUSHARE, output=OUTPUT_FORMAT.DATAFRAME)
+ market=MARKET_TYPE.STOCK_CN, source=DATASOURCE.TUSHARE, output=OUTPUT_FORMAT.DATAFRAME)
- #print(type(list_from_tushare))
+ # print(type(list_from_tushare))
print((list_from_tushare))
-
for idx_df in df_from_Tdx.index:
for dict_elem_in_list in list_from_tushare:
@@ -49,7 +47,7 @@ def test_QA_quotation(self):
open_price1 = df_from_Tdx.loc[idx_df, "open"]
open_price2 = dict_elem_in_list["open"]
- self.assertEqual( open_price1 , open_price2 )
+ self.assertEqual(open_price1, open_price2)
close_price1 = df_from_Tdx.loc[idx_df, "close"]
close_price2 = dict_elem_in_list["close"]
@@ -67,10 +65,11 @@ def test_QA_quotation(self):
vol_price2 = dict_elem_in_list["volume"]
self.assertEqual(vol_price1, vol_price2)
- print("数据对比 tushare 和 tdx 日期 %s 数据一致 ❤️"%idx_df)
+ print("数据对比 tushare 和 tdx 日期 %s 数据一致 ❤️" % idx_df)
print("--------------- ")
pass
+
if __name__ == '__main__':
- unittest.main()
\ No newline at end of file
+ unittest.main()
diff --git a/QUANTAXIS_Test/QAFetch_Test/QAQuery_Advance_Test.py b/QUANTAXIS_Test/QAFetch_Test/QAQuery_Advance_Test.py
index 74a289219..c19abb9c9 100644
--- a/QUANTAXIS_Test/QAFetch_Test/QAQuery_Advance_Test.py
+++ b/QUANTAXIS_Test/QAFetch_Test/QAQuery_Advance_Test.py
@@ -9,16 +9,17 @@
import unittest
+
import fnmatch
import os
-from QUANTAXIS import *;
import sqlite3
import tushare as QATs
-#from QUANTAXIS.QASU.main import (QA_SU_save_stock_list)
-from QUANTAXIS.QASU.main import (select_save_engine)
+from QUANTAXIS import *
+#from QUANTAXIS.QASU.main import (QA_SU_save_stock_list)
+from QUANTAXIS.QASU.main import select_save_engine
class Test_Query_Advance(unittest.TestCase):
@@ -37,7 +38,7 @@ class Test_Query_Advance(unittest.TestCase):
'''
def setUp(self):
- #
+
# mac or linux 可以使用wine 来运行 ， 需要指定字符集防止乱码
# env LC_ALL=zh_CN.UTF-8 wine instjd_1000.exe
#
@@ -62,6 +63,14 @@ def setUp(self):
print(" ")
+ #self.stock_list = QA_fetch_stock_list_adv()
+ # print(self.stock_list)
+
+ #stock_list_items = [item for item in collections.find()];
+ # if stock_list_items is None:
+ # print("💢 Error QA_fetch_stock_list_adv return None, maybe the DATABASE.stock_list have no DATA!")
+ # return None
+
pass
def tearDown(self):
@@ -69,10 +78,18 @@ def tearDown(self):
pass
def test_QA_fetch_stock_min_adv(self):
+
# dataStruct = QA_fetch_stock_min_adv(start='2018-05-28 00:00:00',code = '300439')
# print("获取1分钟数据")
# print(dataStruct)
# #dataStruct.show()
+
+ dataStruct = QA_fetch_stock_min_adv(
+ start='2018-05-28 00:00:00', code='300439')
+ print("获取1分钟数据")
+ print(dataStruct)
+ # dataStruct.show()
+
pass
def test_001(self):
@@ -81,12 +98,16 @@ def test_001(self):
# df = QATs.get_stock_basics()
# print(df)
# print("-----------------------------------------------------------------------")
+
#data = QA_fetch_get_stock_list(package = "tdx")
+
+ data = QA_fetch_get_stock_list(package="tdx")
+
# print(data)
# print("-----------------------------------------------------------------------")
#engine = select_save_engine(engine="Tushare")
- #engine.QA_SU_save_stock_list(client=DATABASE)
+ # engine.QA_SU_save_stock_list(client=DATABASE)
# date = str(datetime.date.today())
# date_stamp = QA_util_date_stamp(date)
@@ -97,5 +118,9 @@ def test_001(self):
# coll = client.stock_list
# coll.insert({'date': date, 'date_stamp': date_stamp,
# 'stock': {'code': data}})
+
#return list(df.index)
- pass
\ No newline at end of file
+ pass
+
+
+
diff --git a/QUANTAXIS_Test/QAFetch_Test/QAQuery_Test.py b/QUANTAXIS_Test/QAFetch_Test/QAQuery_Test.py
index 469468d63..2143eac78 100644
--- a/QUANTAXIS_Test/QAFetch_Test/QAQuery_Test.py
+++ b/QUANTAXIS_Test/QAFetch_Test/QAQuery_Test.py
@@ -4,15 +4,16 @@
from QUANTAXIS.QAFetch.QAQuery import (QA_fetch_stock_basic_info_tushare, QA_fetch_stock_to_market_date)
from QUANTAXIS.QASU.save_to_db_fields_description import quantaxis__db_description
+
class QAQuery_test(unittest.TestCase):
def test_QA_fetch_stock_basic_info_tusare(self):
items = QA_fetch_stock_basic_info_tushare()
print(type(items))
print(len(items))
print("查找 半导体 主营的 股票：")
- #print(items)
+ # print(items)
for i in items:
- #print(i)
+ # print(i)
if i['industry'] == '半导体':
print(i)
@@ -24,4 +25,4 @@ def test_QA_fetch_stock_to_market_date(self):
def test_quantaxis__db_description(self):
- print(quantaxis__db_description)
\ No newline at end of file
+ print(quantaxis__db_description)
diff --git a/QUANTAXIS_Test/QAFetch_Test/QATdx_Test.py b/QUANTAXIS_Test/QAFetch_Test/QATdx_Test.py
index ca626c432..6fe49f79d 100644
--- a/QUANTAXIS_Test/QAFetch_Test/QATdx_Test.py
+++ b/QUANTAXIS_Test/QAFetch_Test/QATdx_Test.py
@@ -1,18 +1,17 @@
+import datetime
+import os
+import socket
+import struct
+import sys
+import time
import unittest
-
import urllib.request
import zipfile
-import struct
-import time
-import os
-import sys
-import socket
-import pandas as pd
-import datetime
+import pandas as pd
-from QUANTAXIS.QAFetch import (QATdx );
-from QUANTAXIS.QAUtil import (QADate, QADate_trade );
+from QUANTAXIS.QAFetch import QATdx
+from QUANTAXIS.QAUtil import QADate, QADate_trade
#from QUANTAXIS import *;
@@ -23,96 +22,101 @@
# 对比数据
#
+
def Schedule(a, b, c):
-# '''''
-# a:已经下载的数据块
-# b:数据块的大小
-# c:远程文件的大小
-# '''
-# per = 100.0 * a * b / c
-# if per > 100:
-# per = 100
-# print
-# '下载进度%.2f%%' % per
+ # '''''
+ # a:已经下载的数据块
+ # b:数据块的大小
+ # c:远程文件的大小
+ # '''
+ # per = 100.0 * a * b / c
+ # if per > 100:
+ # per = 100
+ # print
+ # '下载进度%.2f%%' % per
pass
+
class Test_QA_Fetch(unittest.TestCase):
- #下载最近5天的K线数据
+ # 下载最近5天的K线数据
def setUp(self):
print('Get the free stock data from wstock.net ')
- #获取 下载路径的特殊值
+ # 获取 下载路径的特殊值
url = "http://www.wstock.net/wstock/inc/wstockV2.js"
- #print(url)
+ # print(url)
user_agent = 'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'
})
response = urllib.request.urlopen(req)
content = response.read()
- #print(type(content))
- #print(content)
- strings = content.decode("utf-8","ignore")
- string_lines = strings.split("\r\n");
+ # print(type(content))
+ # print(content)
+ strings = content.decode("utf-8", "ignore")
+ string_lines = strings.split("\r\n")
for line in string_lines:
- #print(type(line))
- #var M_SHSZ_PATH="4T1IWc2q";
+ # print(type(line))
+ # var M_SHSZ_PATH="4T1IWc2q";
if line.startswith("var M_SHSZ_PATH"):
print(line)
- variables = line.split("=");
+ variables = line.split("=")
string_of_path_v = variables[1]
- print(string_of_path_v);
- real_value_password = string_of_path_v.lstrip("\"");
- real_value_password = real_value_password.rstrip("\";");
- print(real_value_password);
+ print(string_of_path_v)
+ real_value_password = string_of_path_v.lstrip("\"")
+ real_value_password = real_value_password.rstrip("\";")
+ print(real_value_password)
- #http://www.wstock.net/wstock/download/4T1IWc2q/wss0502r.zip
+ # http://www.wstock.net/wstock/download/4T1IWc2q/wss0502r.zip
url2 = "http://www.wstock.net/wstock/download/%s/wss%sr.zip"
now = QADate.QA_util_time_now()
- str_from_today = '%04d-%02d-%02d'%(now.year, now.month, now.day);
+ str_from_today = '%04d-%02d-%02d' % (now.year, now.month, now.day)
toDayIsTradeDay = QADate_trade.QA_util_if_trade(str_from_today)
nowTimeIsTrading = QADate_trade.QA_util_if_tradetime(now)
- #todo QADate_trade.QA_util_if_before_today_tradetime 返回是否是一天开盘前
- #todo QADate_trade.QA_util_if_after_today_tradetime 返回是否是一天收盘后
- _time1 = datetime.datetime.strptime(str(now)[0:19], '%Y-%m-%d %H:%M:%S')
- if _time1.hour in [0, 1,2,3,4,5,6,7,8,9,10]:
+ # todo QADate_trade.QA_util_if_before_today_tradetime 返回是否是一天开盘前
+ # todo QADate_trade.QA_util_if_after_today_tradetime 返回是否是一天收盘后
+ _time1 = datetime.datetime.strptime(
+ str(now)[0:19], '%Y-%m-%d %H:%M:%S')
+ if _time1.hour in [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]:
toDayIsBeforeOpen = True
else:
toDayIsBeforeOpen = False
prev_trade_day = ""
if toDayIsTradeDay == False:
- prev_trade_day = QADate_trade.QA_util_get_real_date(str_from_today); # 今天不是交易日，往前移一天
- print("非交易日，下载前一天交易日 %s" % (prev_trade_day));
+ prev_trade_day = QADate_trade.QA_util_get_real_date(
+ str_from_today) # 今天不是交易日，往前移一天
+ print("非交易日，下载前一天交易日 %s" % (prev_trade_day))
elif nowTimeIsTrading == True or toDayIsBeforeOpen == True:
- prev_trade_day = QADate_trade.QA_util_get_real_date(str_from_today); # 当天交易时段无法下载今天的数据，往前移一天
- prev_trade_day = QADate_trade.QA_util_get_last_day(prev_trade_day);
- print("交易时间，下载前一天交易日 %s" % (prev_trade_day));
+ prev_trade_day = QADate_trade.QA_util_get_real_date(
+ str_from_today) # 当天交易时段无法下载今天的数据，往前移一天
+ prev_trade_day = QADate_trade.QA_util_get_last_day(prev_trade_day)
+ print("交易时间，下载前一天交易日 %s" % (prev_trade_day))
else:
- prev_trade_day = str_from_today;
- print("下载今天交易日 %s" % (prev_trade_day));
-
+ prev_trade_day = str_from_today
+ print("下载今天交易日 %s" % (prev_trade_day))
self.test_day_k_line_dates = []
self.test_day_k_line_dad_file_name = []
- dayLeft = 1 # 下载5天之前的数据
+ dayLeft = 1 # 下载5天之前的数据
while dayLeft > 0:
str_month = prev_trade_day[5:7]
- str_day = prev_trade_day[8:10]
- url_data = "http://www.wstock.net/wstock/download/%s/wss%s%sr.zip" % (real_value_password, str_month, str_day)
- print(url_data);
+ str_day = prev_trade_day[8:10]
+ url_data = "http://www.wstock.net/wstock/download/%s/wss%s%sr.zip" % (
+ real_value_password, str_month, str_day)
+ print(url_data)
- #下载文件
- local_file_name = "wss%s%sr.zip"%(str_month,str_day)
- urllib.request.urlretrieve(url_data, local_file_name,Schedule)
+ # 下载文件
+ local_file_name = "wss%s%sr.zip" % (str_month, str_day)
+ urllib.request.urlretrieve(url_data, local_file_name, Schedule)
- #解压缩文件
+ # 解压缩文件
zfile = zipfile.ZipFile(local_file_name, 'r')
for filename in zfile.namelist():
data = zfile.read(filename)
@@ -121,19 +125,18 @@ def setUp(self):
file.close()
self.test_day_k_line_dad_file_name.append(filename)
- strTestDate = "%04d-%s-%s"%(now.year,str_month,str_day);
+ strTestDate = "%04d-%s-%s" % (now.year, str_month, str_day)
self.test_day_k_line_dates.append(strTestDate)
- #删除zip文件 ， 防止误删，暂不实现
+ # 删除zip文件 ， 防止误删，暂不实现
prev_trade_day = QADate_trade.QA_util_get_last_day(prev_trade_day)
dayLeft = dayLeft - 1
pass
-
def tearDown(self):
print('tearDown...')
- #删除dad文件，防止误删，暂不实现
+ # 删除dad文件，防止误删，暂不实现
'''
/*
@@ -158,38 +161,34 @@ def tearDown(self):
'''
- def extraDataFromDAD(self, local_file_name):
+ def extraDataFromDAD(self, local_file_name):
- #解析dad文件
+ # 解析dad文件
df = pd.DataFrame(columns=["stock_name", "date", "open", "high", "low", "close", "volumn", "turn"],
index=["stock_code"])
-
fileDad = open(local_file_name, 'rb', buffering=0)
- fileDad.seek(0x0, 0x0);
+ fileDad.seek(0x0, 0x0)
fileAllBytes = fileDad.readall()
fileDad.close()
#first4Bytes = fileDad.read(0x04);
- first4Bytes = fileAllBytes[0x00: 0x04];
-
+ first4Bytes = fileAllBytes[0x00: 0x04]
if first4Bytes[0] == 0x8c and first4Bytes[1] == 0x19 and first4Bytes[2] == 0xfc and first4Bytes[3] == 0x33:
-
- #fileDad.seek(0x08);
+ # fileDad.seek(0x08);
#byteNumberOfStock = fileDad.read(0x04);
byteNumberOfStock = fileAllBytes[0x08:0x08+0x04]
longNumberOfStock = struct.unpack('<L', byteNumberOfStock)
print("共有")
- print(longNumberOfStock);
-
+ print(longNumberOfStock)
for iStockIndex in range(0, longNumberOfStock[0]):
- #修改， fix 没有输出
+ # 修改， fix 没有输出
#print('共%d个数据 ， 已经读取%d 读取数据中 :%10.8s%s' % (longNumberOfStock[0], iStockIndex, str(iStockIndex / longNumberOfStock[0]), '%'))
i = round((iStockIndex / longNumberOfStock[0]) * 100.0)
@@ -199,19 +198,20 @@ def extraDataFromDAD(self, local_file_name):
#fileDad.seek(0x10 + iStockIndex * 4 * 0x10);
#aStockData = fileDad.read(0x10 * 4);
- aStockData = fileAllBytes[0x10 + iStockIndex * 4 * 0x10: 0x10 + iStockIndex * 4 * 0x10 + 0x10 * 4]
+ aStockData = fileAllBytes[0x10 + iStockIndex * 4 *
+ 0x10: 0x10 + iStockIndex * 4 * 0x10 + 0x10 * 4]
if aStockData[0] == 0xFF and aStockData[1] == 0xFF and aStockData[2] == 0xFF and aStockData[3] == 0xFF:
codeNameByte = aStockData[4:0x10]
- #print(codeNameByte)
- strCodeName = codeNameByte.decode('ascii');
- #print(strCodeName);
+ # print(codeNameByte)
+ strCodeName = codeNameByte.decode('ascii')
+ # print(strCodeName);
stockNameByte = aStockData[0x14: 0x20]
- #print(stockNameByte);
- strStockName = stockNameByte.decode('gbk');
- #print(strStockName);
+ # print(stockNameByte);
+ strStockName = stockNameByte.decode('gbk')
+ # print(strStockName);
stockTime = aStockData[0x20: 0x24]
stockTimeNumber = struct.unpack('<L', stockTime)
@@ -220,81 +220,94 @@ def extraDataFromDAD(self, local_file_name):
#dt = time.strftime("%Y-%m-%d %H:%M:%S", time_local)
dt = time.strftime("%Y-%m-%d", time_local)
- #print(dt);
-
- i=1
- byte_stock_open = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 2;
- byte_stock_high = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 3;
- byte_stock_low = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 4;
- byte_stock_close = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 5;
- byte_stock_volume = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 6;
- byte_stock_turn = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 7;
+ # print(dt);
+
+ i = 1
+ byte_stock_open = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 2
+ byte_stock_high = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 3
+ byte_stock_low = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)]
+ i = 4
+ byte_stock_close = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 5
+ byte_stock_volume = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 6
+ byte_stock_turn = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 7
v1 = struct.unpack('<f', byte_stock_open)
- stock_open = v1[0];
+ stock_open = v1[0]
v1 = struct.unpack('<f', byte_stock_close)
- stock_close = v1[0];
+ stock_close = v1[0]
v1 = struct.unpack('<f', byte_stock_low)
- stock_low = v1[0];
+ stock_low = v1[0]
v1 = struct.unpack('<f', byte_stock_high)
- stock_high = v1[0];
+ stock_high = v1[0]
v1 = struct.unpack('<f', byte_stock_volume)
- stock_volume = v1[0];
+ stock_volume = v1[0]
v1 = struct.unpack('<f', byte_stock_turn)
- stock_turn = v1[0];
+ stock_turn = v1[0]
#print("%f %f %f %f %f %f "%(stock_open, stock_close, stock_high,stock_low, stock_volume, stock_turn))
- #print("------")
+ # print("------")
df.index.astype(str)
- df.loc[strCodeName[0:8]] = [strStockName, dt, stock_open, stock_high, stock_low, stock_close, stock_volume, stock_turn]
+ df.loc[strCodeName[0:8]] = [strStockName, dt, stock_open,
+ stock_high, stock_low, stock_close, stock_volume, stock_turn]
return df
def test_QA_fetch_get_stock_day(self):
+ # 读取 dad 日线数据
+ df_from_dad = self.extraDataFromDAD(
+ self.test_day_k_line_dad_file_name[0])
- #读取 dad 日线数据
- df_from_dad = self.extraDataFromDAD(self.test_day_k_line_dad_file_name[0])
-
- df_from_tdx = QATdx.QA_fetch_get_stock_day(code="600000",start_date= self.test_day_k_line_dates[0], end_date = self.test_day_k_line_dates[0])
+ df_from_tdx = QATdx.QA_fetch_get_stock_day(
+ code="600000", start_date=self.test_day_k_line_dates[0], end_date=self.test_day_k_line_dates[0])
for idx_from_df_tdx in df_from_tdx.index:
print(idx_from_df_tdx)
- #for idx_from_df_dad in df_from_dad.index:
+ # for idx_from_df_dad in df_from_dad.index:
- open_price1 = df_from_tdx.loc[idx_from_df_tdx, "open"]
+ open_price1 = df_from_tdx.loc[idx_from_df_tdx, "open"]
close_price1 = df_from_tdx.loc[idx_from_df_tdx, "close"]
- high_price1 = df_from_tdx.loc[idx_from_df_tdx, "high"]
- low_price1 = df_from_tdx.loc[idx_from_df_tdx, "low"]
- volume1 = df_from_tdx.loc[idx_from_df_tdx, "vol"]
- amount1 = df_from_tdx.loc[idx_from_df_tdx, "amount"]
- date1 = df_from_tdx.loc[idx_from_df_tdx, "date"]
+ high_price1 = df_from_tdx.loc[idx_from_df_tdx, "high"]
+ low_price1 = df_from_tdx.loc[idx_from_df_tdx, "low"]
+ volume1 = df_from_tdx.loc[idx_from_df_tdx, "vol"]
+ amount1 = df_from_tdx.loc[idx_from_df_tdx, "amount"]
+ date1 = df_from_tdx.loc[idx_from_df_tdx, "date"]
- #fix here 构造和 QA_fetch_get_stock_day 返回一样类型的 dataframe 直接比较
+ # fix here 构造和 QA_fetch_get_stock_day 返回一样类型的 dataframe 直接比较
- open_price2 = df_from_dad.loc["SH600000", "open"]
+ open_price2 = df_from_dad.loc["SH600000", "open"]
close_price2 = df_from_dad.loc["SH600000", "close"]
- high_price2 = df_from_dad.loc["SH600000", "high"]
- low_price2 = df_from_dad.loc["SH600000", "low"]
- volume2 = df_from_dad.loc["SH600000", "volumn"]
- amount2 = df_from_dad.loc["SH600000", "turn"]
- date2 = df_from_dad.loc["SH600000", "date"]
-
- self.assertEqual( open_price1 , round(open_price2,2) )
- self.assertEqual( close_price1 , round(close_price2,2) )
- self.assertEqual( high_price1 , round(high_price2,2) )
- self.assertEqual( low_price1 , round(low_price2,2) )
+ high_price2 = df_from_dad.loc["SH600000", "high"]
+ low_price2 = df_from_dad.loc["SH600000", "low"]
+ volume2 = df_from_dad.loc["SH600000", "volumn"]
+ amount2 = df_from_dad.loc["SH600000", "turn"]
+ date2 = df_from_dad.loc["SH600000", "date"]
+
+ self.assertEqual(open_price1, round(open_price2, 2))
+ self.assertEqual(close_price1, round(close_price2, 2))
+ self.assertEqual(high_price1, round(high_price2, 2))
+ self.assertEqual(low_price1, round(low_price2, 2))
# 成交量和成交额 总是有误差 ？？
#self.assertEqual( volume1 , volume2 )
#self.assertEqual( amount1 , amount2)
- self.assertEqual( date1 , date2 )
+ self.assertEqual(date1, date2)
df_from_tdx = QATdx.QA_fetch_get_stock_day(code="300439", start_date=self.test_day_k_line_dates[0],
end_date=self.test_day_k_line_dates[0])
@@ -302,22 +315,22 @@ def test_QA_fetch_get_stock_day(self):
print(idx_from_df_tdx)
# for idx_from_df_dad in df_from_dad.index:
- open_price1 = df_from_tdx.loc[idx_from_df_tdx, "open"]
+ open_price1 = df_from_tdx.loc[idx_from_df_tdx, "open"]
close_price1 = df_from_tdx.loc[idx_from_df_tdx, "close"]
- high_price1 = df_from_tdx.loc[idx_from_df_tdx, "high"]
- low_price1 = df_from_tdx.loc[idx_from_df_tdx, "low"]
- volume1 = df_from_tdx.loc[idx_from_df_tdx, "vol"]
- amount1 = df_from_tdx.loc[idx_from_df_tdx, "amount"]
- date1 = df_from_tdx.loc[idx_from_df_tdx, "date"]
+ high_price1 = df_from_tdx.loc[idx_from_df_tdx, "high"]
+ low_price1 = df_from_tdx.loc[idx_from_df_tdx, "low"]
+ volume1 = df_from_tdx.loc[idx_from_df_tdx, "vol"]
+ amount1 = df_from_tdx.loc[idx_from_df_tdx, "amount"]
+ date1 = df_from_tdx.loc[idx_from_df_tdx, "date"]
# fix here 构造和 QA_fetch_get_stock_day 返回一样类型的 dataframe 直接比较
- open_price2 = df_from_dad.loc["SZ300439", "open"]
+ open_price2 = df_from_dad.loc["SZ300439", "open"]
close_price2 = df_from_dad.loc["SZ300439", "close"]
- high_price2 = df_from_dad.loc["SZ300439", "high"]
- low_price2 = df_from_dad.loc["SZ300439", "low"]
- volume2 = df_from_dad.loc["SZ300439", "volumn"]
- amount2 = df_from_dad.loc["SZ300439", "turn"]
- date2 = df_from_dad.loc["SZ300439", "date"]
+ high_price2 = df_from_dad.loc["SZ300439", "high"]
+ low_price2 = df_from_dad.loc["SZ300439", "low"]
+ volume2 = df_from_dad.loc["SZ300439", "volumn"]
+ amount2 = df_from_dad.loc["SZ300439", "turn"]
+ date2 = df_from_dad.loc["SZ300439", "date"]
self.assertEqual(open_price1, round(open_price2, 2))
self.assertEqual(close_price1, round(close_price2, 2))
@@ -328,8 +341,8 @@ def test_QA_fetch_get_stock_day(self):
# self.assertEqual( amount1 , amount2)
self.assertEqual(date1, date2)
- #print(idx_from_df_dad)
+ # print(idx_from_df_dad)
#print(type(open_price1), type(close_price1) ,type(high_price1), type(low_price1), type(volume1), type(amount1), type(date1))
pass
- pass
\ No newline at end of file
+ pass
diff --git a/QUANTAXIS_Test/QAMarket_Test/QA_Order_Test.py b/QUANTAXIS_Test/QAMarket_Test/QA_Order_Test.py
index c13d1fbd0..1bb2c17c2 100644
--- a/QUANTAXIS_Test/QAMarket_Test/QA_Order_Test.py
+++ b/QUANTAXIS_Test/QAMarket_Test/QA_Order_Test.py
@@ -12,4 +12,3 @@ def tearDown(self):
def testQAOrder(self):
print(" test QAOrder !")
-
diff --git a/QUANTAXIS_Test/QAUtil_Test/QADate_trade_Test.py b/QUANTAXIS_Test/QAUtil_Test/QADate_trade_Test.py
index 018c7b4ec..880a46cbb 100644
--- a/QUANTAXIS_Test/QAUtil_Test/QADate_trade_Test.py
+++ b/QUANTAXIS_Test/QAUtil_Test/QADate_trade_Test.py
@@ -1,20 +1,18 @@
import datetime
import unittest
-
-from QUANTAXIS.QAFetch import (QATdx );
-from QUANTAXIS.QAUtil import (QADate, QADate_trade );
+from QUANTAXIS.QAFetch import QATdx
+from QUANTAXIS.QAUtil import QADate, QADate_trade
class Test_QA_Date_trade(unittest.TestCase):
def test_QA_util_if_trade(self):
-
now = QADate.QA_util_time_now()
str_from_today = '%04d-%02d-%02d' % (now.year, now.month, now.day)
nDayLeft = 1000
- while nDayLeft>0:
+ while nDayLeft > 0:
toDayIsTradeDay = QADate_trade.QA_util_if_trade(str_from_today)
realTradeDay = QADate_trade.QA_util_get_real_date(str_from_today)
@@ -22,13 +20,15 @@ def test_QA_util_if_trade(self):
QADate.QA_util_date_valid(realTradeDay)
if toDayIsTradeDay == False:
- prev_trade_day = QADate_trade.QA_util_get_last_day(str_from_today,-1)
- realTradeDay = QADate_trade.QA_util_get_real_date(prev_trade_day)
- self.assertEquals(realTradeDay , toDayIsTradeDay)
+ prev_trade_day = QADate_trade.QA_util_get_last_day(
+ str_from_today, -1)
+ realTradeDay = QADate_trade.QA_util_get_real_date(
+ prev_trade_day)
+ self.assertEquals(realTradeDay, toDayIsTradeDay)
else:
- self.assertEquals(realTradeDay , realTradeDay)
+ self.assertEquals(realTradeDay, realTradeDay)
str_from_today = QADate_trade.QA_util_get_last_day(str_from_today)
nDayLeft = nDayLeft - 1
- print("okok")
\ No newline at end of file
+ print("okok")
diff --git a/QUANTAXIS_Test/QAUtil_Test/QATest_None.py b/QUANTAXIS_Test/QAUtil_Test/QATest_None.py
index ba8f0b395..c42eea8da 100644
--- a/QUANTAXIS_Test/QAUtil_Test/QATest_None.py
+++ b/QUANTAXIS_Test/QAUtil_Test/QATest_None.py
@@ -1,22 +1,24 @@
-import unittest
-
-import sys
-import time
import datetime
import random
+import sys
+import time
+import unittest
-#from QUANTAXIS.QAFetch import (QATdx );
-from QUANTAXIS.QAUtil import (QADate, QADate_trade );
-
+import matplotlib as mpl
#from pandas import Series
import pandas as pd
-import matplotlib as mpl
-#学习 lru_cache
+#from QUANTAXIS.QAFetch import (QATdx );
+from QUANTAXIS.QAUtil import QADate, QADate_trade
+
+# 学习 lru_cache
+
+
class MyCache:
'''
'''
+
def __init__(self):
'''constructor'''
self.cache = {}
@@ -39,7 +41,8 @@ def update(self, key, value):
'''
if key not in self.cache and len(self.cache) >= self.max_cache_size:
self.remove_oldest()
- self.cache[key] = { 'date_accessed': datetime.datetime.now(), 'value': value}
+ self.cache[key] = {
+ 'date_accessed': datetime.datetime.now(), 'value': value}
def remove_oldest(self):
"""
@@ -50,10 +53,10 @@ def remove_oldest(self):
for key in self.cache:
if oldest_entry == None:
oldest_entry = key
- print('assign oldest_entry key',oldest_entry)
+ print('assign oldest_entry key', oldest_entry)
elif self.cache[key]['date_accessed'] < self.cache[oldest_entry]['date_accessed']:
oldest_entry = key
- print('delete oles key',oldest_entry)
+ print('delete oles key', oldest_entry)
self.cache.pop(oldest_entry)
@@ -65,12 +68,11 @@ def size(self):
return len(self.cache)
-
class Test_QA_None(unittest.TestCase):
def test_LRU(self):
# 测试缓存
- keys = ['test', 'red', 'fox', 'fence', 'junk', \
+ keys = ['test', 'red', 'fox', 'fence', 'junk',
'other7', 'alpha8', 'bravo9', 'cal10', 'devo11', 'ele12',
'other1', 'alpha2', 'bravo3', 'cal4', 'devo5', 'ele6',
]
@@ -85,11 +87,10 @@ def test_LRU(self):
print("#%s iterations, #%s cached entries" % (i + 1, cache.size))
pass
-
- def fab(self,max):
+ def fab(self, max):
n, a, b = 0, 0, 1
while n < max:
- print (b)
+ print(b)
a, b = b, a + b
n = n + 1
@@ -99,10 +100,9 @@ def test_Generator(self):
def test_None(self):
print("----------------------------")
- obj = pd.Series([3,4,-2,2])
+ obj = pd.Series([3, 4, -2, 2])
print(obj)
-
pd.Series([1, 23, 4]).plot()
#now = QADate.QA_util_time_now()
@@ -120,26 +120,25 @@ def test_None(self):
# sys.stdout.flush()
# pass
-
def test_make_Series(self):
- #demo the series
- obj = pd.Series([1,2,3,4])
+ # demo the series
+ obj = pd.Series([1, 2, 3, 4])
print(obj.index)
print(obj.values)
print("-------------------------------------------")
- obj2 = pd.Series([1,2,3,4], index=['a','b','c','d'])
+ obj2 = pd.Series([1, 2, 3, 4], index=['a', 'b', 'c', 'd'])
print(obj2)
print(obj2.index)
print(obj2.values)
print("-------------------------------------------")
# another way to think about a Series is as a fixed-length, ordered dict, as it is a mapping of index value to data values.
# it can be substituted into many functions that expect a dict
- sdata = { 'Ohio': 3000, 'Texas': 71000, 'Oregon': 1600, 'Utah': 5000}
+ sdata = {'Ohio': 3000, 'Texas': 71000, 'Oregon': 1600, 'Utah': 5000}
obj4 = pd.Series(sdata)
print(obj4)
print("-------------------------------------------")
state = ['California', 'Ohio', 'Oregon', 'Taxes']
- obj4 = pd.Series(sdata, index = state)
+ obj4 = pd.Series(sdata, index=state)
print(obj4)
print("-------------------------------------------")
print(pd.isnull(obj4))
@@ -147,9 +146,9 @@ def test_make_Series(self):
print(pd.notnull(obj4))
print("-------------------------------------------")
- obj_01 = pd.Series([1,2,3,4])
- obj_o2 = pd.Series([11,22,33,44], index=['aa','bb','cc','dd'])
- obj_03 = pd.Series([55,66,77,88], index=['bb','cc','dd','ee'])
+ obj_01 = pd.Series([1, 2, 3, 4])
+ obj_o2 = pd.Series([11, 22, 33, 44], index=['aa', 'bb', 'cc', 'dd'])
+ obj_03 = pd.Series([55, 66, 77, 88], index=['bb', 'cc', 'dd', 'ee'])
print(obj_01)
print(obj_o2)
print("-------------------------------------------")
@@ -157,20 +156,20 @@ def test_make_Series(self):
print("-------------------------------------------")
print(obj_o2 + obj_03)
-
def test_make_dataframe(self):
# the dataframe has both a row and column index
dict_data = {
'state': ['Ohio', 'Ohio', 'Ohio', 'Nevada', 'Nevada'],
'year': [2000, 2001, 2002, 2001, 2002],
- 'pop': [1.4,1.7,3.6,2.4,2.9]
+ 'pop': [1.4, 1.7, 3.6, 2.4, 2.9]
}
frame = pd.DataFrame(dict_data)
print(frame)
print("-------------------------------------------")
- frame2 = pd.DataFrame(dict_data, columns=['year','state','pop', 'area'])
+ frame2 = pd.DataFrame(dict_data, columns=[
+ 'year', 'state', 'pop', 'area'])
print(frame2)
print("-------------------------------------------")
@@ -189,8 +188,7 @@ def test_make_dataframe(self):
print(frame2.T)
print("-------------------------------------------")
- pop = { 'Nevada' : {2001: 2.4, 2002: 2.9}, "Ohio": {2000: 1.4,2000: 1.7, 2002: 3.4}}
+ pop = {'Nevada': {2001: 2.4, 2002: 2.9},
+ "Ohio": {2000: 1.4, 2000: 1.7, 2002: 3.4}}
frame3 = pd.DataFrame(pop)
print(frame3)
-
-
diff --git a/QUANTAXIS_Test/QAUtil_Test/QATester_nothing.py b/QUANTAXIS_Test/QAUtil_Test/QATester_nothing.py
index 677737cf5..d8ed51486 100644
--- a/QUANTAXIS_Test/QAUtil_Test/QATester_nothing.py
+++ b/QUANTAXIS_Test/QAUtil_Test/QATester_nothing.py
@@ -1,13 +1,12 @@
+import datetime
+import struct
+import time
import unittest
-
#from urllib import request
import urllib
import urllib.request
-import datetime;
-import struct;
-import time
-import pandas as pd
+import pandas as pd
#from QUANTAXIS import QUANTAXIS as QA
@@ -33,6 +32,8 @@
字节串转16进制表示,固定两个字符表示: str(binascii.b2a_hex(b'\x01\x0212'))[2:-1] ==> 01023132
字节串转16进制数组: [hex(x) for x in bytes(b'\x01\x0212')] ==> ['0x1', '0x2', '0x31', '0x32']
'''
+
+
class QA_Test(unittest.TestCase):
def setUp(self):
today = datetime.date.today()
@@ -40,7 +41,7 @@ def setUp(self):
print(today.year)
print(today.month)
print(today.day)
- str = "%04d-%02d-%02d"%(today.year,today.month,today.day)
+ str = "%04d-%02d-%02d" % (today.year, today.month, today.day)
print(str)
pass
@@ -59,27 +60,27 @@ def nottest_QA(self):
index=["code"])
if first4Bytes[0] == 0x8c and first4Bytes[1] == 0x19 and first4Bytes[2] == 0xfc and first4Bytes[3] == 0x33:
- fileDad.seek(0x08);
- byteNumberOfStock = fileDad.read(0x04);
+ fileDad.seek(0x08)
+ byteNumberOfStock = fileDad.read(0x04)
longNumberOfStock = struct.unpack('<L', byteNumberOfStock)
- #print(longNumberOfStock);
+ # print(longNumberOfStock);
for iStockIndex in range(0, longNumberOfStock[0]):
- fileDad.seek(0x10 + iStockIndex * 4 * 0x10);
+ fileDad.seek(0x10 + iStockIndex * 4 * 0x10)
- aStockData = fileDad.read(0x10 * 4);
+ aStockData = fileDad.read(0x10 * 4)
if aStockData[0] == 0xFF and aStockData[1] == 0xFF and aStockData[2] == 0xFF and aStockData[3] == 0xFF:
codeNameByte = aStockData[4:0x10]
- #print(codeNameByte)
- strCodeName = codeNameByte.decode('gbk');
- #print(strCodeName);
+ # print(codeNameByte)
+ strCodeName = codeNameByte.decode('gbk')
+ # print(strCodeName);
stockNameByte = aStockData[0x14: 0x20]
- #print(stockNameByte);
- strStockName = stockNameByte.decode('gbk');
- #print(strStockName);
+ # print(stockNameByte);
+ strStockName = stockNameByte.decode('gbk')
+ # print(strStockName);
stockTime = aStockData[0x20: 0x24]
stockTimeNumber = struct.unpack('<L', stockTime)
@@ -88,49 +89,60 @@ def nottest_QA(self):
#dt = time.strftime("%Y-%m-%d %H:%M:%S", time_local)
dt = time.strftime("%Y-%m-%d", time_local)
- #print(dt);
-
- i=1
- byte_stock_open = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 2;
- byte_stock_close = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 3;
- byte_stock_low = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 4;
- byte_stock_high = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 5;
- byte_stock_volume = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 6;
- byte_stock_turn = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)];i = 7;
+ # print(dt);
+
+ i = 1
+ byte_stock_open = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 2
+ byte_stock_close = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 3
+ byte_stock_low = aStockData[0x20+(i * 4): 0x20+((i+1) * 4)]
+ i = 4
+ byte_stock_high = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 5
+ byte_stock_volume = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 6
+ byte_stock_turn = aStockData[0x20 +
+ (i * 4): 0x20+((i+1) * 4)]
+ i = 7
v1 = struct.unpack('<f', byte_stock_open)
- stock_open = v1[0];
+ stock_open = v1[0]
v1 = struct.unpack('<f', byte_stock_close)
- stock_close = v1[0];
+ stock_close = v1[0]
v1 = struct.unpack('<f', byte_stock_low)
- stock_low = v1[0];
+ stock_low = v1[0]
v1 = struct.unpack('<f', byte_stock_high)
- stock_high = v1[0];
+ stock_high = v1[0]
v1 = struct.unpack('<f', byte_stock_volume)
- stock_volume = v1[0];
+ stock_volume = v1[0]
v1 = struct.unpack('<f', byte_stock_turn)
- stock_turn = v1[0];
+ stock_turn = v1[0]
#print("%f %f %f %f %f %f "%(stock_open, stock_close, stock_high,stock_low, stock_volume, stock_turn))
- #print("------")
+ # print("------")
df.index.astype(str)
- df.loc[strCodeName] = [strStockName, dt, stock_open, stock_close, stock_low, stock_high, stock_volume, stock_turn]
+ df.loc[strCodeName] = [strStockName, dt, stock_open,
+ stock_close, stock_low, stock_high, stock_volume, stock_turn]
pass
pass
fileDad.close()
-
print(df)
- return df;
+ return df
- pass
\ No newline at end of file
+ pass