modules/CreateTickerDatabase.py - master - malkinte/TA_moex

modules/CreateTickerDatabase.py
253 строки12 KB
Teresa Malkin
first commit
30 июн 2025, 20:46
30 июн 2025, 20:46e22bcdc
import sqlite3
from functions import *

class TickerDatabase:
    def __init__(self, db_path: str, tickers_file_path: str):
        self.db_path = db_path
        self.tickers_file_path = tickers_file_path
        self.connection = None
        self.cursor = None

    def connect(self):
        '''Устанавливает соединение с базой данных'''
        self.connection = sqlite3.connect(self.db_path)
        self.cursor = self.connection.cursor()

    def ticker_exists(self, table: str, ticker: str) -> bool:
        '''Проверяет, существует ли тикер в базе данных'''
        self.cursor.execute(f'SELECT 1 FROM {table} WHERE ticker = ?', ([ticker]))
        return self.cursor.fetchone() is not None

    def create_tickers_preferences_table(self):
        '''Создаёт таблицу tickers_preferences, если она не существует'''
        self.cursor.execute(
            '''
            CREATE TABLE IF NOT EXISTS tickers_preferences (
                id INTEGER PRIMARY KEY AUTOINCREMENT, ticker TEXT UNIQUE, name TEXT,
                "complex product" INTEGER, "only long" INTEGER,
                "profit target" DOUBLE, "risk control" DOUBLE
                )
            '''
        )

    def create_filters_table(self):
        '''Создаёт таблицу filters, если она не существует'''
        self.cursor.execute(
            '''
            CREATE TABLE IF NOT EXISTS filters (
                id INTEGER PRIMARY KEY AUTOINCREMENT, ticker TEXT UNIQUE,
                "Filter_1" INTEGER,     "Filter_2" INTEGER,  "Filter_3" INTEGER,
                "Filter_4" INTEGER,     "Filter_5" INTEGER,  "Filter_6" INTEGER,
                "Filter_7" INTEGER,     "Filter_8" INTEGER,  "Filter_9" INTEGER,
                "Filter_10" INTEGER,    "Filter_11" INTEGER, "Filter_12" INTEGER,
                "Filter_13" INTEGER,    "Filter_14" INTEGER, "Filter_15" INTEGER,
                "Filter_16" INTEGER,    "Filter_17" INTEGER, "Filter_18" INTEGER,
                "Filter_19" INTEGER,    "Filter_20" INTEGER
                )
            '''
        )

    def create_ticks_history_table(self):
        '''Создаёт таблицу ticks_history, если она не существует'''
        self.cursor.execute(
            '''
            CREATE TABLE IF NOT EXISTS ticks_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                Date DATE,
                Ticker TEXT,
                Open DECIMAL(15, 6),
                High DECIMAL(15, 6),
                Low DECIMAL(15, 6),
                Close DECIMAL(15, 6),
                Volume INTEGER
                --UNIQUE(Date, Ticker)
                )
            '''
        )

    def fill_tickers_preferences_table(self):
        '''Загружает данные из файла в базу данных'''
        try:
            with open(self.tickers_file_path, 'r', newline='\n', encoding='UTF-8') as tickers_file:
                lines = sorted(tickers_file.read().split('\n'))

                for idx, line in enumerate(lines, 0):
                    if not line.strip():
                        continue

                    try:
                        # Удаляем лишние пробелы и разбиваем строку
                        parts = line.strip().split(',')
                        parts = [part.strip() for part in parts]  # Удаляем пробелы вокруг каждого элемента

                        # Проверяем, что у нас достаточно элементов
                        if len(parts) < 6:
                            print(f"Ошибка в строке {idx}: недостаточно значений. Пропускаем.")
                            continue

                        # Извлекаем значения
                        ticker, name, complex_product, only_long, profit_target, loss_target, *filters = parts

                        # Проверяет, существует ли тикер в базе данных
                        if self.ticker_exists('tickers_preferences', ticker):
                            continue

                        # Убедитесь, что у вас 20 фильтров
                        if len(filters) != 20:
                            print(f"Ошибка в строке {idx}: недостаточно фильтров. Пропускаем.")
                            continue

                        # Подготовка к выполнению запроса
                        values_to_insert = (ticker, name, complex_product, only_long, profit_target, loss_target)
                        print(f"Вставка в строке {idx}: {values_to_insert}")  # Вывод значений перед вставкой

                        self.cursor.execute(
                            '''
                            INSERT INTO tickers_preferences (
                                ticker,      name,            "complex product",
                                "only long", "profit target", "risk control"
                                ) 
                            VALUES (?, ?, ?, ?, ?, ?)
                            ''',
                            values_to_insert
                        )

                    except ValueError as err1:
                        print(f"Ошибка в строке {idx}: '{line}'. Пропускаем. Ошибка: {err1}")

            self.connection.commit()
        except FileNotFoundError:
            print(f"Файл не найден: {self.tickers_file_path}")
            raise
        except Exception as err2:
            print(f"Ошибка при загрузке данных: {err2}")
            raise

    def fill_filters_table(self):
        '''Загружает данные из файла в базу данных'''
        try:
            with open(self.tickers_file_path, 'r', newline='\n', encoding='UTF-8') as tickers_file:
                lines = sorted(tickers_file.read().split('\n'))

                for idx, line in enumerate(lines, 0):
                    if not line.strip():
                        continue

                    try:
                        # Удаляем лишние пробелы и разбиваем строку
                        parts = line.strip().split(',')
                        parts = [part.strip() for part in parts]  # Удаляем пробелы вокруг каждого элемента

                        # Проверяем, что у нас достаточно элементов
                        if len(parts) < 6:
                            print(f"Ошибка в строке {idx}: недостаточно значений. Пропускаем.")
                            continue

                        # Извлекаем значения
                        ticker, name, complex_product, only_long, profit_target, loss_target, *filters = parts

                        # Проверяет, существует ли тикер в базе данных
                        if self.ticker_exists('filters', ticker):
                            continue

                        # Убедитесь, что у вас 20 фильтров
                        if len(filters) != 20:
                            print(f"Ошибка в строке {idx}: недостаточно фильтров. Пропускаем.")
                            continue

                        # Подготовка к выполнению запроса
                        values_to_insert = (ticker, *filters)
                        print(f"Вставка в строке {idx}: {values_to_insert}")  # Вывод значений перед вставкой

                        self.cursor.execute(
                            '''
                            INSERT INTO filters (
                                ticker, 
                                "Filter_1",  "Filter_2",      "Filter_3",
                                "Filter_4",  "Filter_5",      "Filter_6",
                                "Filter_7",  "Filter_8",      "Filter_9",
                                "Filter_10", "Filter_11",     "Filter_12",
                                "Filter_13", "Filter_14",     "Filter_15",
                                "Filter_16", "Filter_17",     "Filter_18",
                                "Filter_19", "Filter_20"
                                ) 
                            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                            ''',
                            (ticker, *filters)
                        )

                    except ValueError as err1:
                        print(f"Ошибка в строке {idx}: '{line}'. Пропускаем. Ошибка: {err1}")

            self.connection.commit()
        except FileNotFoundError:
            print(f"Файл не найден: {self.tickers_file_path}")
            raise
        except Exception as err2:
            print(f"Ошибка при загрузке данных: {err2}")
            raise

    def fill_ticks_history_table(self, month_shift=300, day_shift=2):
        range_of_request = get_range_of_request(month_shift=month_shift, day_shift=day_shift)
        self.cursor.execute('SELECT * FROM tickers_preferences')
        tickers = self.cursor.fetchall()

        for ticker in tickers:
            self.cursor.execute('SELECT Date FROM ticks_history WHERE ticker = ?', ([ticker[1]]))
            dates = [date[0] for date in self.cursor.fetchall()]
            # print(dates[:-1])  # Удаляем последнюю запись?
            
            ticker += range_of_request
            print(f"Вставка: {ticker[:2]}: ", end='')

            ticks, columns = get_request_data(ticker, '24')
            entry_count = 0
            for tick in ticks:
                if date_exists(tick[7], dates):
                    continue
                
                self.cursor.execute(
                    '''
                    INSERT INTO ticks_history (
                        Date, Ticker, Open, High, Low, Close, Volume
                        ) 
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    ''',
                    (tick[7], ticker[1], tick[0], tick[2], tick[3], tick[1], tick[5])
                )
                entry_count += 1
            print('Внесено записей:', entry_count)
            
        self.connection.commit()

    def close(self):
        '''Закрывает соединение с базой данных'''
        if self.connection:
            self.connection.close()
            self.connection = None
            self.cursor = None

    def execute(self):
        '''Основной метод для загрузки тикеров и фильтров в базу данных'''
        try:
            self.connect()
            self.create_tickers_preferences_table()
            self.create_filters_table()
            self.create_ticks_history_table()
            self.fill_tickers_preferences_table()
            self.fill_filters_table()
            self.fill_ticks_history_table()
            print("Данные успешно загружены в базу.")
        finally:
            self.close()


if __name__ == '__main__':
    try:
        db = TickerDatabase(
            db_path           = '../Data/tickers_database.sqlite',
            tickers_file_path = '../Tickers_specification.txt'
        )
        db.execute()
    except Exception as err0:
        print(f"Программа завершена с ошибкой: {err0}")