Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
wiseplat
GitHub Repository: wiseplat/python-code
 Path: blob/master/ invest-robot-contest_tinkoff-invest-volume-analysis-robot-master/tests/test_profile_touch_strategy.py
5927 views
1
import csv

2
import datetime

3
import logging

4


5
import pandas as pd

6


7
from services.order_service import OrderService

8
from strategies.profile_touch_strategy import ProfileTouchStrategy

9
from utils.logger import init_logging

10
from utils.parse_util import parse_date

11
from utils.format_util import fixed_float

12
from utils.strategy_util import apply_frame_type

13


14
pd.options.display.max_columns = None

15
pd.options.display.max_rows = None

16
pd.options.display.width = None

17


18
init_logging()

19
logger = logging.getLogger(__name__)

20


21


22
class TestProfileTouchStrategy:

23
    def __init__(self, instrument_name, file_path):

24
        self.instrument_name = instrument_name

25
        self.file_path = file_path

26


27
        self.df = pd.DataFrame(columns=["figi", "direction", "price", "quantity", "time"])

28
        self.df = apply_frame_type(self.df)

29


30
        self.profile_touch_strategy = ProfileTouchStrategy(instrument_name)

31
        self.profile_touch_strategy.start()

32


33
        self.order_service = OrderService()

34
        self.order_service.start()

35


36
    def run(self):

37
        test_start_time = datetime.datetime.now()

38
        logger.info("анализ истории: %s", self.file_path)

39


40
        with open(self.file_path, newline='') as file:

41
            reader = csv.DictReader(file, delimiter=",")

42
            for row in reader:

43
                figi = row["figi"]

44
                price = float(row["price"])

45
                time = parse_date(row["time"])

46


47
                processed_trade_df = pd.DataFrame.from_records([

48
                    {

49
                        "figi": figi,

50
                        "direction": row["direction"],

51
                        "price": price,

52
                        "quantity": row["quantity"],

53
                        "time": pd.to_datetime(str(time), utc=True),

54
                    }

55
                ])

56
                self.order_service.processed_orders(self.instrument_name, price, time)

57


58
                processed_trade_df = apply_frame_type(processed_trade_df)

59
                orders = self.profile_touch_strategy.analyze(processed_trade_df)

60
                if orders is not None:

61
                    for order in orders:

62
                        self.order_service.create_order(order)

63


64
        # после завершения анализа перестраиваю график, т.к. закрытие торгов не совпадает целому часу

65
        # например 15:59:59.230333+00:00

66
        self.profile_touch_strategy.calculate_clusters()

67
        self.order_service.write_statistics()

68


69
        test_end_time = datetime.datetime.now()

70
        total_test_time = (test_end_time - test_start_time).total_seconds() / 60

71
        logger.info("анализ завершен")

72
        logger.info("время тестирования: %s мин.", fixed_float(total_test_time))

73


74


75
if __name__ == "__main__":

76
    usd_histories = {"name": "USD000UTSTOM",

77
                     "files": ["./../data/USD000UTSTOM-20220504.csv", "./../data/USD000UTSTOM-20220505.csv",

78
                               "./../data/USD000UTSTOM-20220506.csv", "./../data/USD000UTSTOM-20220511.csv",

79
                               "./../data/USD000UTSTOM-20220512.csv", "./../data/USD000UTSTOM-20220513.csv",

80
                               "./../data/USD000UTSTOM-20220516.csv", "./../data/USD000UTSTOM-20220517.csv",

81
                               "./../data/USD000UTSTOM-20220518.csv", "./../data/USD000UTSTOM-20220519.csv",

82
                               "./../data/USD000UTSTOM-20220520.csv"]}

83


84
    sber_histories = {"name": "SBER",

85
                      "files": ["./../data/SBER-20220504.csv", "./../data/SBER-20220505.csv",

86
                                "./../data/SBER-20220506.csv", "./../data/SBER-20220511.csv",

87
                                "./../data/SBER-20220512.csv", "./../data/SBER-20220513.csv",

88
                                "./../data/SBER-20220516.csv", "./../data/SBER-20220517.csv",

89
                                "./../data/SBER-20220518.csv", "./../data/SBER-20220519.csv",

90
                                "./../data/SBER-20220520.csv"]}

91


92
    gaz_histories = {"name": "GAZP",

93
                     "files": ["./../data/GAZP-20220504.csv", "./../data/GAZP-20220505.csv",

94
                               "./../data/GAZP-20220506.csv", "./../data/GAZP-20220511.csv",

95
                               "./../data/GAZP-20220512.csv", "./../data/GAZP-20220513.csv",

96
                               "./../data/GAZP-20220516.csv", "./../data/GAZP-20220517.csv",

97
                               "./../data/GAZP-20220518.csv", "./../data/GAZP-20220519.csv",

98
                               "./../data/GAZP-20220520.csv", ]}

99


100
    for history in [usd_histories, sber_histories, gaz_histories]:

101
        for file_path in history["files"]:

102
            tester = TestProfileTouchStrategy(history["name"], file_path)

103
            tester.run()

104


105