37 // Trade - это сделка о покупке/продаже по символу ценной бумаги.
38 type Trade struct {
39     Time   time.Time
40     Symbol string
41     Price  float64
42     IsBuy  bool
43 }

24     schemaSQL = `
25 CREATE TABLE IF NOT EXISTS trades (
26     time TIMESTAMP,
27     symbol VARCHAR(32),
28     price FLOAT,
29     buy BOOLEAN
30 );
31
32 CREATE INDEX IF NOT EXISTS trades_time ON trades(time);
33 CREATE INDEX IF NOT EXISTS trades_symbol ON trades(symbol);
34 `

16     insertSQL = `
17 INSERT INTO trades (
18     time, symbol, price, buy
19 ) VALUES (
20     ?, ?, ?, ?
21 )
22 `

45 // DB - это база данных биржевых торгов.
46 type DB struct {
47     sql    *sql.DB
48     stmt   *sql.Stmt
49     buffer []Trade
50 }

53 // NewDB создает Trades для управления торговлей акциями в базе данных SQLite.
Этот API не является потокобезопасным.
54 func NewDB(dbFile string) (*DB, error) {
55     sqlDB, err := sql.Open("sqlite3", dbFile)
56     if err != nil {
57         return nil, err
58     }
59
60     if _, err = sqlDB.Exec(schemaSQL); err != nil {
61         return nil, err
62     }
63
64     stmt, err := sqlDB.Prepare(insertSQL)
65     if err != nil {
66         return nil, err
67     }
68
69     db := DB{
70         sql:    sqlDB,
71         stmt:   stmt,
72         buffer: make([]Trade, 0, 1024),
73     }
74     return &db, nil
75 }

77 // Add сохраняет сделку в буфер. Как только буфер заполняется,
сделки вносятся в базу данных.
79 func (db *DB) Add(trade Trade) error {
80     if len(db.buffer) == cap(db.buffer) {
81         return errors.New("trades buffer is full")
82     }
83
84     db.buffer = append(db.buffer, trade)
85     if len(db.buffer) == cap(db.buffer) {
86         if err := db.Flush(); err != nil {
87             return fmt.Errorf("unable to flush trades: %w", err)
88         }
89     }
90
91     return nil
92 }

94  // Flush вносит ждущие обработки сделки в базу данных.
95  func (db *DB) Flush() error {
96      tx, err := db.sql.Begin()
97      if err != nil {
98          return err
99      }
100
101     for _, trade := range db.buffer {
102         _, err := tx.Stmt(db.stmt).Exec(trade.Time, trade.Symbol, trade.Price, trade.IsBuy)
103         if err != nil {
104             tx.Rollback()
105             return err
106         }
107     }
108
109     db.buffer = db.buffer[:0]
110     return tx.Commit()
111 }

113 // Close вносит (посредством Flush) все сделки в базу данных и предотвращает любую торговлю в будущем.
114 func (db *DB) Close() error {
115     defer func() {
116         db.stmt.Close()
117         db.sql.Close()
118     }()
119
120     if err := db.Flush(); err != nil {
121         return err
122     }
123
124     return nil
125 }

5 // Ваши пакеты main и test требуют эти импорты, чтобы пакет sql был правильно инициализирован.
6 // _ "github.com/mattn/go-sqlite3"
7
8 import (
9     "database/sql"
10     "errors"
11     "fmt"
12     "time"
13 )

66 func ExampleDB() {
67     dbFile := "/tmp/db-test" + time.Now().Format(time.RFC3339)
68     db, err := trades.NewDB(dbFile)
69     if err != nil {
70         fmt.Println("ERROR: create -", err)
71         return
72     }
73     defer db.Close()
74
75     const count = 10_000
76     for i := 0; i < count; i++ {
77         trade := trades.Trade{
78             Time:   time.Now(),
79             Symbol: "AAPL",
80             Price:  rand.Float64() * 200,
81             IsBuy:  i%2 == 0,
82         }
83         if err := db.Add(trade); err != nil {
84             fmt.Println("ERROR: insert - ", err)
85             return
86         }
87     }
88
89     fmt.Printf("inserted %d records\n", count)
90     // Вывод:
91     // inserted 10000 records
92 }

02 import sqlite3
03 from contextlib import closing
04 from datetime import datetime
05
06 import pandas as pd

08 select_sql = """
09 SELECT * FROM trades
10 WHERE time >= ? AND time <= ?
11 """

14 def load_trades(db_file, start_time, end_time):
15     """Загружаем сделки из db_file за заданный временной диапазон."""
16     conn = sqlite3.connect(db_file)
17     with closing(conn) as db:
18         df = pd.read_sql(select_sql, db, params=(start_time, end_time))
19
20     # Мы не можем использовать здесь detect_types=sqlite3.PARSE_DECLTYPES, поскольку Go вставляет часовой пояс, а sqlite3 Python не умеет обрабатывать его.
22     # Смотрите https://bugs.python.org/issue29099# См. Https://bugs.python.org/issue29099
23     df["time"] = pd.to_datetime(df["time"])
24     return df

27 def average_price(df):
28     """Возвращает среднюю цену в df, сгруппированную по (stock, buy)"""
29     return df.groupby(["symbol", "buy"], as_index=False)["price"].mean()

symbol,buy,price
AAPL,0,250.82925665004535
AAPL,1,248.28277375538832
GOOG,0,250.11537993385295
GOOG,1,252.4726772487683
MSFT,0,250.9214212695317
MSFT,1,248.60187022941685
NVDA,0,250.3844763417279
NVDA,1,249.3578146208962