an open quant framework (based on xtquant & qmt mini)
from typing import List, Dict
from datetime import datetime
import os

from open_quant_app.manager.PositionManager import PositionManager
from open_quant_app.model.BarData import BarData
from xtquant import xtconstant
from xtquant.xttype import XtPosition

from loguru import logger
import pandas as pd
import matplotlib.pyplot as plt


class BackPosition:
    def __init__(self, stock_id: str, price: float = 0.0, volume: int = 0):
        self.price = price
        self.avg_price = price
        self.volume = volume
        self.stock_id = stock_id

    def buy(self, price: float, volume: int):
        if volume == 0:
            return
        self.price = price
        self.avg_price = (self.avg_price * self.volume + price * volume) / (self.volume + volume)
        self.volume += volume

    def sell(self, price: float, volume: int) -> float:
        self.price = price
        self.volume -= volume
        return price * volume

    def value(self):
        return self.price * self.volume


class BackAccount:
    def __init__(self, cash: float = 0.0):
        self.initial_cash = cash
        self.cash = cash

    def deposit(self, money: float):
        self.cash += money

    def withdraw(self, money: float):
        self.cash -= money


class BackOrder:
    def __init__(self, order_type: int, price: float, volume: int, stock_id: str, strategy_id: int):
        self.order_type = order_type
        self.price = price
        self.volume = volume
        self.stock_id = stock_id
        self.strategy_id = strategy_id


class BackTester:
    def __init__(self, config: dict, cash: float = 100000, stock_list: List[str] = None):
        self.account = BackAccount(cash)
        self.stock_ids = config['stock']['stock_ids']
        self.stocks: Dict[str, BackPosition] = {}
        for stock_id in self.stocks:
            self.stocks[stock_id] = BackPosition(stock_id)
        self.position_manager = PositionManager(config)
        self.records: List[BackOrder] = []
        self.asset_records: Dict[datetime, float] = {}
        self.bar_data_records: Dict[str, Dict[datetime, BarData]] = {}
        self.begin: datetime = datetime.now()
        self.end: datetime = datetime.now()

    def set_time_seg(self, begin: datetime, end: datetime):
        self.begin = begin
        self.end = end

    # info functions
    def info(self):
        logger.info(f"initial cash = {self.account.cash}")

    def value(self) -> float:
        total_assets = 0
        for stock_id in self.stocks:
            total_assets += self.stocks[stock_id].value()
        total_assets += self.account.cash
        return total_assets

    # record functions
    def record_assets(self, timestamp: datetime):
        self.asset_records[timestamp] = self.value()

    def record_bar_data(self, bar_data: BarData):
        if bar_data.stock_id not in self.bar_data_records:
            self.bar_data_records[bar_data.stock_id] = {}
        self.bar_data_records[bar_data.stock_id][bar_data.timestamp] = bar_data

    # trade sim functions
    def query_position(self, stock_id: str) -> int:
        if stock_id in self.stocks:
            return self.stocks[stock_id].volume
        else:
            return 0

    def close_position(self, stock_id: str):
        if stock_id not in self.stocks:
            return
        position = self.stocks[stock_id]
        money = self.stocks[stock_id].sell(position.price, position.volume)
        self.account.deposit(money)

    def can_buy(self, volume: int, price: float, strategy_id: int) -> bool:
        xt_positions = []
        for stock_id in self.stock_ids[strategy_id]:
            back_position = self.stocks[stock_id]
            # simulate position data
            xt_position = XtPosition('backtester', stock_id, back_position.volume, back_position.volume, 0, 0, 0, 0, 0,
                                     back_position.avg_price)
            xt_positions.append(xt_position)
        total_assets = self.value()
        return not self.position_manager.is_position_limit(xt_positions, strategy_id, volume, price, total_assets)

    def can_sell(self, stock_id: str, volume: int, strategy_id: int) -> bool:
        if stock_id not in self.stocks:
            logger.warning(f"you have 0 position for stock {stock_id}, cannot sell !")
            return False
        else:
            position = self.stocks[stock_id]
            if position.volume < volume:
                logger.warning(f"position available volume = {position.volume} < sell volume {volume}, cancel !")
                return False
            else:
                return True

    def order_stock(self, stock_id: str, order_type: int, volume: int, price: float, strategy_id: int) -> int:
        if stock_id not in self.stocks:
            self.stocks[stock_id] = BackPosition(stock_id)
        if order_type == xtconstant.STOCK_BUY and self.can_buy(volume, price, strategy_id):
            self.stocks[stock_id].buy(price, volume)
            self.account.withdraw(price * volume)
            self.records.append(BackOrder(xtconstant.STOCK_BUY, price, volume, stock_id, strategy_id))
            return 0
        elif order_type == xtconstant.STOCK_SELL and self.can_sell(stock_id, volume, strategy_id):
            money = self.stocks[stock_id].sell(price, volume)
            self.account.deposit(money)
            self.records.append(BackOrder(xtconstant.STOCK_SELL, price, volume, stock_id, strategy_id))
            return 0
        else:
            return -1

    # report functions
    def check_output_dirs(self):
        dirs = ['../output/trade', '../output/bars', '../output/asset']
        for _dir in dirs:
            if not os.path.exists(_dir):
                os.makedirs(_dir)

    def report(self, strategy_id: int, save_as_file: bool = True,
               plot_assets: bool = True, plot_bars: bool = True):
        self.check_output_dirs()
        # log assets & ratio
        logger.critical(
            f"init = {self.account.initial_cash}, curr = {self.value()}, "
            f"ratio = {(self.value() - self.account.initial_cash) / self.account.initial_cash}")
        # save orders
        if not save_as_file:
            for i in range(len(self.records)):
                record = self.records[i]
                logger.info(
                    f"[{i}], stock id = {record.stock_id}, type = {'buy' if record.order_type == xtconstant.STOCK_BUY else 'sell'}"
                    f", price = {record.price}, volume = {record.volume}")
        elif len(self.records) != 0:
            df = pd.DataFrame([vars(record) for record in self.records])
            df[df['strategy_id'] == strategy_id].to_csv(f'../output/trade/strategy-{strategy_id}.csv')
        # plot assets curve
        if plot_assets:
            self.plot_asset_curve()
        if plot_bars:
            self.plot_bars()

    def plot_asset_curve(self):
        timestamps = list(self.asset_records.keys())
        assets = list(self.asset_records.values())

        plt.figure(figsize=(10, 5))
        plt.plot(timestamps, assets)
        plt.title("Datetime vs Pure Assets")
        plt.xlabel("Datetime")
        plt.ylabel("Pure Assets")
        plt.grid(True)
        plt.tight_layout()
        plt.savefig('../output/asset/assets-curve.jpg')
        plt.show()

    def plot_bars(self):
        for stock_id in self.bar_data_records:
            bar_data_dict = self.bar_data_records[stock_id]
            timestamps = list(bar_data_dict.keys())
            bar_data = list(bar_data_dict.values())
            close = [i.close for i in bar_data]

            plt.figure(figsize=(10, 5))
            plt.plot(timestamps, close)
            plt.title(f"Datetime vs Close [{stock_id}]")
            plt.xlabel("Datetime")
            plt.ylabel("Close")
            plt.grid(True)
            plt.tight_layout()
            plt.savefig(f'../output/bars/{stock_id}.jpg')
            plt.show()

open_quant_app/backtest/BackTester.py
openhe-hub/open-quant-app
from datetime import datetime
from typing import Type
from threading import Thread

from loguru import logger
from open_quant_app.trade.Trader import Trader, TradeMode
from open_quant_app.strategy.Strategy import Strategy


class QuantApp:
    def __init__(self, config: dict, strategy: Type[Strategy], mode: TradeMode = TradeMode.MARKET, cash: int = 200000):
        logger.info("=== STRATEGY INIT ===")
        self.config: dict = config
        self.mode: TradeMode = mode
        # init trader
        logger.info("initializing trader module ...")
        if self.mode == TradeMode.MARKET:
            self.trader = Trader(self.config)
        elif self.mode == TradeMode.BACKTEST:
            self.trader = Trader(self.config, mode, cash=cash)
        self.trader.start()
        self.trader.info()
        # init strategies
        logger.info("initializing strategy module ...")
        self.strategy: Type[Strategy] = strategy
        self.strategies = []
        for i in range(len(self.config['stock']['stock_ids'])):
            self.strategies.append(self.strategy(i, self.trader, self.config))
            logger.success(f"strategy-{i} created")

    def run(self, begin: datetime = None, end: datetime = None):
        logger.info("begin main loop")
        if self.mode == TradeMode.MARKET:
            for strategy in self.strategies:
                strategy_thread = Thread(target=strategy.main_loop, args=())
                strategy_thread.start()
        elif self.mode == TradeMode.BACKTEST:
            for strategy in self.strategies:
                strategy_thread = Thread(target=strategy.main_loop_back, args=(begin, end))
                strategy_thread.start()

open_quant_app/core/QuantApp.py
openhe-hub/open-quant-app
import enum
from typing import Dict, List
from datetime import datetime, timedelta

from open_quant_app.trade.Trader import Trader

from loguru import logger


class Order:
    def __init__(self, order_id: int, order_timestamp: datetime):
        self.order_id: int = order_id
        self.order_timestamp = order_timestamp


class OrderTuple:
    def __init__(self, stock_ids: [str], orders: [Order]):
        self.stock_ids: [str] = stock_ids
        self.orders: [Order] = orders
        self.status: OrderStatus = OrderStatus.INIT


class OrderStatus(enum.Enum):
    INIT = 1
    SELL_UNFINISHED = 2
    BUY_UNFINISHED = 3
    FINISHED = 4


class OrderManager:
    def __init__(self, trader: Trader, stock_ids: List[str], delay: float = 1, sliding_point: float = 0.0005):
        self.orders: List[OrderTuple] = []
        self.trader: Trader = trader
        self.stock_ids: List[str] = stock_ids
        self.delay: float = delay
        self.sliding_point: float = sliding_point

    def insert(self, order_tuple: OrderTuple):
        self.orders.append(order_tuple)

    def size(self) -> int:
        return len(self.orders)

    def empty(self) -> bool:
        return self.size() == 0

    def clear_finished(self):
        self.orders = list(filter(lambda order_tuple: order_tuple.status != OrderStatus.FINISHED, self.orders))

    def handle_once(self, order_tuple: OrderTuple) -> OrderTuple:
        return order_tuple

    def handle(self):
        logger.success(f"handle begin. {self.size()} order tuples left")
        for i in range(self.size()):
            self.orders[i] = self.handle_once(self.orders[i])
        self.clear_finished()

        logger.success(f"handle done. {self.size()} order tuples left")

open_quant_app/manager/OrderManager.py
openhe-hub/open-quant-app
import enum
import time
from typing import Dict, List
from datetime import datetime, timedelta

from loguru import logger
import xtquant.xtconstant as xtconstant
import xtquant.xtdata as xt_data
from xtquant.xttype import XtOrder

from open_quant_app.trade.Trader import Trader
from open_quant_app.manager.OrderManager import OrderStatus


class OrderManagerV2:
    def __init__(self, trader: Trader, strategy_id: int, stock_ids: List[str], delay: float = 1,
                 sliding_point: float = 0.0005):
        self.strategy_id: int = strategy_id
        self.trader: Trader = trader
        self.stock_ids: List[str] = stock_ids
        self.delay: float = delay
        self.sliding_point: float = sliding_point
        self.finished: Dict[int, bool] = {}

    def sync_finished_orders(self):
        for order in self.trader.query_orders():
            self.finished[order.order_id] = True

    def if_sliding_point(self, curr_price: float, ref_price: float) -> bool:
        if (abs(curr_price - ref_price) / ref_price) > self.sliding_point:
            logger.error(f"sliding point potential detected: curr = {curr_price}, ref = {ref_price}")
            return True
        else:
            return False

    def get_latest_price(self, stock_id: str) -> dict:
        # get data
        timestamp = datetime.now()
        end_timestamp = (timestamp + timedelta(minutes=1)).strftime("%Y%m%d%H%M%S")
        prev_timestamp = (timestamp + timedelta(minutes=-1)).strftime("%Y%m%d%H%M%S")
        data = xt_data.get_market_data(field_list=['askPrice', 'bidPrice'],
                                       stock_list=[stock_id], period='tick', count=1,
                                       start_time=prev_timestamp, end_time=end_timestamp,
                                       dividend_type='front', fill_data=True)
        # subtract price
        ask_price = data[stock_id]['askPrice'][-1][0] if len(data[stock_id]['askPrice']) != 0 else -1
        bid_price = data[stock_id]['bidPrice'][-1][0] if len(data[stock_id]['bidPrice']) != 0 else -1

        if ask_price == -1 or bid_price == -1:
            logger.critical(f"0 value detected in OrderManagerV2: ask price or bid price")
            return None

        return {
            "ask_price": ask_price,
            "bid_price": bid_price,
        }

    def handle_once(self, order_id: int):
        pass

    def handle(self):
        pass

    def detect(self) -> List[int]:
        pass
open_quant_app/manager/OrderManagerV2.py
openhe-hub/open-quant-app
from typing import List

from loguru import logger
from xtquant.xttype import XtPosition


class PositionManager:
    def __init__(self, config: dict):
        self.position_avg_mode = config['assets']['position_avg_mode']
        self.max_cash = config['assets']['max_cash']
        self.num = len(config['stock']['stock_ids'])
        self.positions = []
        if self.position_avg_mode:
            for i in range(self.num):
                self.positions.append(1.0 / self.num)
        else:
            self.positions = config['assets']['positions']
            if len(self.positions) != self.num:
                logger.error(f"unequal position setting: {self.num} stock tuples, but {len(self.positions)} "
                             f"position limits")

    def is_position_limit(self, position_data: List[XtPosition], strategy_id: int, volume: int, price: float,
                          total_assets: float) -> bool:
        stock_assets = 0
        for position in position_data:
            if position is None:
                return False
            stock_assets += position.open_price * (position.can_use_volume + position.frozen_volume)
        if stock_assets + price * volume >= self.positions[strategy_id] * min(total_assets, self.max_cash):
            logger.warning(f"potential stock asset = {stock_assets + price * volume}, position limit "
                           f"= {self.positions[strategy_id]}, cannot buy more !")
            return True
        else:
            return False

open_quant_app/manager/PositionManager.py
openhe-hub/open-quant-app
from datetime import datetime

class BarData:
    def __init__(self, stock_id: str, timestamp: datetime, close: float, strategy_id: int = 0):
        self.stock_id: str = stock_id
        self.timestamp: datetime = timestamp
        self.close: float = close
        self.strategy_id: int = strategy_id
open_quant_app/model/BarData.py
openhe-hub/open-quant-app
from datetime import datetime
import time
from typing import List, Dict

import xtquant.xtdata as xtdata
from open_quant_app.trade.Trader import Trader
from open_quant_app.utils.FixedQueue import FixedQueue
from open_quant_app.manager.OrderManager import OrderManager
from open_quant_app.utils.TimeUtils import TimeUtils
from open_quant_app.model.BarData import BarData

from loguru import logger


class StrategyData:
    def __init__(self, strategy_id: int = 0, timestamp: datetime = datetime.now()):
        self.strategy_id: int = strategy_id
        self.timestamp: datetime = timestamp

    def empty(self) -> bool:
        return False


class Strategy:
    def __init__(self, strategy_id: int, trader: Trader, config: dict = None):
        self.strategy_id: int = strategy_id
        self.trader: Trader = trader
        self.stock_ids: [str] = config['stock']['stock_ids'][self.strategy_id]
        self.period: float = config['strategy']['periods']
        self.order_manager: OrderManager = OrderManager(trader, self.stock_ids)
        self.records: FixedQueue[StrategyData] = FixedQueue(config['ui']['record_length'])

    def subscribe_quotes(self, period_list: [str]):
        for stock_id in self.stock_ids:
            for period in period_list:
                xtdata.subscribe_quote(stock_id, period=period, callback=self.on_data_callback)

    def on_data_callback(self, data_cbk):
        pass

    def get_bar_data(self) -> Dict[str, BarData]:
        return {}

    def exec(self, timestamp: datetime = None) -> StrategyData:
        return StrategyData()

    def exec_before(self, timestamp: datetime = datetime.now()):
        pass

    def exec_after(self, timestamp: datetime = datetime.now()):
        pass

    def main_loop(self):
        self.exec_before()
        timestamp = datetime.now()
        while not TimeUtils.after_afternoon_end(timestamp):
            if TimeUtils.judge_trade_time(timestamp):
                record = self.exec(timestamp)
                if not record.empty():
                    self.records.append(record)
            else:
                logger.warning(f"curr timestamp =  {timestamp} not in trade time!")
            time.sleep(self.period)
        self.exec_after()

    def main_loop_back(self, start: datetime, end: datetime = datetime.now()):
        self.trader.back_tester.set_time_seg(start, end)
        curr_timestamp = start
        while curr_timestamp < end:
            record = self.exec(curr_timestamp)
            if not record.empty():
                self.records.append(record)
            # record asset
            self.trader.back_tester.record_assets(curr_timestamp)
            # record bar data
            bar_data_dict = self.get_bar_data()
            for stock_id in bar_data_dict:
                self.trader.back_tester.record_bar_data(bar_data_dict[stock_id])

            curr_timestamp = TimeUtils.next_trade_timestamp(curr_timestamp, self.period)
        self.trader.back_tester.report(self.strategy_id, save_as_file=True)

open_quant_app/strategy/Strategy.py
openhe-hub/open-quant-app
from xtquant.xttrader import XtQuantTraderCallback
from xtquant.xttype import XtOrder, XtAsset, XtOrderError, XtOrderResponse, XtPosition, XtCancelError, \
    XtAccountStatus, XtTrade

from loguru import logger


class CommonXtQuantTraderCallback(XtQuantTraderCallback):
    def on_disconnected(self):
        logger.error("Warning: connection lost!")

    def on_stock_order(self, order: XtOrder):
        logger.info("on order callback")
        logger.info(order.stock_code, order.order_status, order.order_sysid)

    def on_stock_asset(self, asset: XtAsset):
        logger.info("on asset callback")
        logger.info(asset.account_id, asset.cash, asset.total_asset)

    def on_stock_trade(self, trade: XtTrade):
        logger.info("on trade callback")
        logger.info(trade.account_id, trade.stock_code, trade.order_id)

    def on_stock_position(self, position: XtPosition):
        logger.info("on position callback")
        logger.info(position.stock_code, position.volume)

    def on_order_error(self, order_error: XtOrderError):
        logger.info("on order_error callback")
        logger.info(order_error.order_id, order_error.error_id, order_error.error_msg)

    def on_cancel_error(self, cancel_error: XtCancelError):
        logger.info("on cancel_error callback")
        logger.info(cancel_error.order_id, cancel_error.error_id, cancel_error.error_msg)

    def on_order_stock_async_response(self, response: XtOrderResponse):
        logger.info("on_order_stock_async_response")
        logger.info(response.account_id, response.order_id, response.seq)

    def on_account_status(self, status: XtAccountStatus):
        logger.info("on_account_status")
        logger.info(status.account_id, status.account_type, status.status)

open_quant_app/trade/CommonTradeCallback.py
openhe-hub/open-quant-app
import enum
from typing import List
import math

from open_quant_app.manager.PositionManager import PositionManager
from xtquant.xttrader import XtQuantTrader
from xtquant.xttype import StockAccount, XtOrder, XtAsset, XtPosition, XtTrade
from xtquant import xtconstant
from open_quant_app.trade.CommonTradeCallback import CommonXtQuantTraderCallback
from open_quant_app.backtest.BackTester import BackTester
from open_quant_app.utils.OrderFilter import OrderFilter
from open_quant_app.utils.LogUtils import LogUtils

from loguru import logger
import numpy as np


class TradeMode(enum.Enum):
    MARKET = 1
    BACKTEST = 2


class Trader:
    def __init__(self, config: dict = None, mode: TradeMode = TradeMode.MARKET, cash: float = 100000):
        self.env_path = config['trade']['env_path']
        self.session_id = config['trade']['session_id']
        self.account_id = config['trade']['account_id']
        self.stock_ids = config['stock']['stock_ids']
        self.max_cash = config['assets']['max_cash']
        self.mode = mode

        self.xt_trader = XtQuantTrader(self.env_path, self.session_id)
        self.account = StockAccount(self.account_id)
        self.callback = CommonXtQuantTraderCallback()
        self.back_tester = BackTester(config, cash)
        self.position_manager = PositionManager(config)
        self.log_utils = LogUtils(config)

    def start(self):
        # start trade thread
        self.xt_trader.start()
        connection_result = self.xt_trader.connect()
        if connection_result != 0:
            logger.error("connect to QMT MINI failed !")
        else:
            logger.success("connect to QMT MINI success !")
        # subscribe trade acc
        subscribe_result = self.xt_trader.subscribe(self.account)
        if subscribe_result != 0:
            logger.error(f"subscribe to account id = {self.account_id} failed !")
        else:
            logger.success(f"subscribe to account id = {self.account_id} success !")

    def info(self):
        logger.info("begin checking basic status")
        if self.mode == TradeMode.MARKET:
            # check assets
            asset = self.query_stock_asset()
            logger.info("you are in MARKET MODE")
            logger.info(f"account id: {asset.account_id}")
            logger.info(f"total asset: {asset.total_asset}")
            logger.info(f"cash: {asset.cash}")
            logger.info(f"market value: {asset.market_value}")
            logger.info(f"frozen cash: {asset.frozen_cash}")
        elif self.mode == TradeMode.BACKTEST:
            logger.info("you are in BACKTEST MODE")
            self.back_tester.info()

    def order_stock(self, stock_id: str, order_type: int, volume: int, price: float, strategy_id: int,
                    price_type: int = xtconstant.FIX_PRICE, strategy_name: str = '', comment: str = '') -> int:
        if volume == 0:
            return 0
        order_id = -1
        if self.mode == TradeMode.MARKET:
            self.log_utils.log(f"stock_id={stock_id}, strategy_id={strategy_id}, vol={volume}, price={price}, "
                               f"type={'buy' if order_type == xtconstant.STOCK_BUY else 'sell'}, "
                               f"check={self.can_buy(volume, price, strategy_id) if order_type == xtconstant.STOCK_BUY else self.can_sell(stock_id, volume, strategy_id)}, "
                               f"reason={self.get_reason_can_sell(stock_id, volume) if order_type == xtconstant.STOCK_SELL else self.get_reason_can_buy(volume, price, strategy_id)}")

            if order_type == xtconstant.STOCK_BUY and self.can_buy(volume, price, strategy_id):
                order_id = self.xt_trader.order_stock(self.account, stock_id, order_type, volume
                                                      , price_type, price, strategy_name, comment)
            elif order_type == xtconstant.STOCK_SELL and self.can_sell(stock_id, volume, strategy_id):
                order_id = self.xt_trader.order_stock(self.account, stock_id, order_type, volume
                                                      , price_type, price, strategy_name, comment)
        elif self.mode == TradeMode.BACKTEST:
            order_id = self.back_tester.order_stock(stock_id, order_type, volume, price, strategy_id)
        if order_id != -1:
            logger.success(
                f"trading: {'buy' if order_type == xtconstant.STOCK_BUY else 'sell'} {stock_id} volume = {volume}"
                f", price = {price}")
        return order_id

    def cancel_order_stock(self, order_id: int) -> int:
        ret = self.xt_trader.cancel_order_stock(self.account, order_id)
        if ret != 0:
            print(f"Err: cancel order id = {order_id} failed !")
        return ret

    def clear_position(self, stock_id: str, price: float, strategy_id: int) -> int:
        if self.mode == TradeMode.MARKET:
            # TODO: FINISH THIS FEAT
            return -1
        elif self.mode == TradeMode.BACKTEST:
            total_volume = self.back_tester.query_position(stock_id)
            return self.back_tester.order_stock(stock_id, xtconstant.STOCK_SELL, total_volume, price, strategy_id)

    def query_orders(self) -> List[XtOrder]:
        orders = self.xt_trader.query_stock_orders(self.account)
        return orders

    def query_order_by_id(self, order_id) -> XtOrder:
        order = self.xt_trader.query_stock_order(self.account, order_id)
        return order

    def query_trades(self) -> List[XtTrade]:
        trades = self.xt_trader.query_stock_trades(self.account)
        return trades

    def query_positions(self) -> List[XtPosition]:
        positions = self.xt_trader.query_stock_positions(self.account)
        return positions

    def query_position_by_stock(self, stock_id: str) -> XtPosition:
        position = self.xt_trader.query_stock_position(self.account, stock_id)
        return position

    def query_stock_asset(self) -> XtAsset:
        asset = self.xt_trader.query_stock_asset(self.account)
        return asset

    def close(self):
        self.xt_trader.unsubscribe(self.account)
        self.xt_trader.stop()

    def can_sell(self, stock_id: str, volume: int, strategy_id: int) -> bool:
        position = self.query_position_by_stock(stock_id)
        if position is None:
            logger.warning(f"you have 0 position for stock {stock_id}, cannot sell !")
            return False
        elif volume > position.can_use_volume:
            logger.warning(f"position available volume = {position.can_use_volume} < sell volume {volume}, cancel !")
            return False
        else:
            return True

    def max_can_sell(self, stock_id: str, strategy_id: int, volume: int) -> int:
        can_use_vol = self.query_position_by_stock(stock_id).can_use_volume
        if self.query_position_by_stock(stock_id).can_use_volume < volume:
            logger.warning(f"cannot sell {volume}, choose max sell volume: {can_use_vol}")
            return can_use_vol
        else:
            return volume

    def get_reason_can_sell(self, stock_id: str, volume: int) -> str:
        position = self.query_position_by_stock(stock_id)
        if position is None:
            return f"None position on {stock_id}"
        elif volume > position.can_use_volume:
            return f"Not enough position of {stock_id}, vol={volume}, available={position.can_use_volume}"
        else:
            return f"can sell"

    def can_buy(self, volume: int, price: float, strategy_id: int) -> bool:
        # query position
        position_data = []
        stock_id_tuple = self.stock_ids[strategy_id]
        for stock_id in stock_id_tuple:
            position_data.append(self.query_position_by_stock(stock_id))
        # query assets
        total_assets = self.query_stock_asset().total_asset
        return not self.position_manager.is_position_limit(position_data, strategy_id, volume, price, total_assets)

    def max_can_buy(self, volume, price, strategy_id: int) -> int:
        # query position
        stock_id_tuple = self.stock_ids[strategy_id]
        position_val = 0.0
        for stock_id in stock_id_tuple:
            position = self.query_position_by_stock(stock_id)
            if position is None:
                continue
            # use open_price, because avg price may be 0
            position_val += (position.can_use_volume + position.frozen_volume) * position.open_price
        # query order
        order_val = 0.0
        orders = OrderFilter.filter(self.query_orders(), stock_id_tuple)
        orders = OrderFilter.filter_order_status(orders, [xtconstant.ORDER_CANCELED, xtconstant.ORDER_SUCCEEDED,
                                                          xtconstant.ORDER_JUNK])
        for order in orders:
            order_val += order.order_volume * order.price * (-1 if order.order_type == xtconstant.STOCK_SELL else 1)
        # query asset
        if self.query_stock_asset() is None:
            return 0
        total_asset = self.query_stock_asset().total_asset
        total_asset = min(total_asset, self.max_cash) * self.position_manager.positions[strategy_id]
        # check
        if price == 0 or volume == 0 or np.isnan((total_asset - position_val - order_val) / price / 10):
            return 0
        # calc max
        can_buy_vol = max(math.floor((total_asset - position_val - order_val) / price / 10), 0)
        if volume > can_buy_vol:
            logger.critical(f"id = {strategy_id}: cannot buy {volume}, choose max buy volume: {can_buy_vol}")
            return can_buy_vol
        else:
            return volume

    def get_reason_can_buy(self, volume: int, price: float, strategy_id: int) -> str:
        # query position
        stock_id_tuple = self.stock_ids[strategy_id]
        position_val = 0.0
        for stock_id in stock_id_tuple:
            position = self.query_position_by_stock(stock_id)
            if position is None:
                continue
            # use open_price, because avg price may be 0
            position_val += (position.can_use_volume + position.frozen_volume) * position.open_price
        # query order
        order_val = 0.0
        for order in self.query_orders():
            if order.stock_code in stock_id_tuple:
                order_val += order.order_volume * order.price * (-1 if order.order_type == xtconstant.STOCK_SELL else 1)
        # query asset
        if self.query_stock_asset() is None:
            return "query asset data failed"
        total_asset = self.query_stock_asset().total_asset
        total_asset = min(total_asset, self.max_cash) * self.position_manager.positions[strategy_id]
        # check
        if price == 0 or volume == 0 or np.isnan((total_asset - position_val - order_val) / price / 10):
            return "query stock data failed"
        # calc max
        can_buy_vol = max(math.floor((total_asset - position_val - order_val) / price / 10), 0)
        if volume > can_buy_vol:
            return f"cannot buy {volume}, choose max buy volume: {can_buy_vol}"
        else:
            return "can buy"

open_quant_app/trade/Trader.py
openhe-hub/open-quant-app
import toml


class ConfigUtils:
    @staticmethod
    def load(path: str) -> dict:
        return toml.load(path)

open_quant_app/utils/ConfigUtils.py
openhe-hub/open-quant-app
from loguru import logger
from datetime import datetime, timedelta

from threading import Thread
import time

import xtquant.xtdata as xt_data


class DownloadUtils:
    @staticmethod
    def on_download_progress(data):
        logger.info(f"downloading... total = {data['total']}, finished = {data['finished']}")

    @staticmethod
    def download_history_data(config: dict, period_list: [str], back_days: int = 15, start_time: str = "",
                              end_time: str = ""):
        # parse
        stock_ids = []
        for i in range(len(config['stock']['stock_ids'])):
            for j in range(len(config['stock']['stock_ids'][i])):
                stock_ids.append(config['stock']['stock_ids'][i][j])
        # use an extra thread to download
        download_thread = None
        for period in period_list:
            if start_time == "" or end_time == "":
                end_timestamp = datetime.now()
                start_timestamp = end_timestamp - timedelta(days=back_days)
                download_thread = Thread(target=xt_data.download_history_data2,
                                         args=(stock_ids, period, start_timestamp.strftime("%Y%m%d")
                                               , end_timestamp.strftime("%Y%m%d"), DownloadUtils.on_download_progress))
            else:
                download_thread = Thread(target=xt_data.download_history_data2,
                                         args=(stock_ids, period, start_time
                                               , end_time, DownloadUtils.on_download_progress))
        download_thread.start()
        time.sleep(5)

open_quant_app/utils/DownloadUtils.py
openhe-hub/open-quant-app
from collections import deque


class FixedQueue(deque):
    def __init__(self, size):
        super().__init__(maxlen=size)

    def append(self, item):
        super().append(item)

open_quant_app/utils/FixedQueue.py
openhe-hub/open-quant-app
import os.path
from datetime import datetime
from loguru import logger


class LogUtils:
    def __init__(self, config: dict):
        self.log_dir = config['log']['output_dir']
        if not os.path.exists(self.log_dir):
            os.makedirs(self.log_dir)

        today_timestamp = datetime.today().strftime('%Y-%m-%d')
        self.log_file_path = os.path.join(self.log_dir, f"{today_timestamp}.log")
        self.loguru_file_path = os.path.join(self.log_dir, f"{today_timestamp}-loguru.log")

    def log(self, message: str):
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        log_message = f"[{timestamp}] {message}\n"
        with open(self.log_file_path, 'a') as file:
            file.write(log_message)

    def open_file_logger(self):
        logger.add(self.loguru_file_path)





open_quant_app/utils/LogUtils.py
openhe-hub/open-quant-app
from typing import List
from xtquant.xttype import XtOrder
import xtquant.xtconstant as xtconst


class OrderFilter:
    @staticmethod
    def filter(orders: List[XtOrder], stock_pool: List[str],
               filter_canceled: bool = True,
               filter_wrong_id: bool = True) -> List[XtOrder]:
        filtered_orders: List[XtOrder] = []
        for order in orders:
            if order.stock_code in stock_pool \
                    and (not filter_canceled or order.order_status != xtconst.ORDER_CANCELED) \
                    and (not filter_wrong_id or order.order_id > 0):
                filtered_orders.append(order)
        return filtered_orders

    @staticmethod
    def filter_order_type(orders: List[XtOrder], order_type: int) -> List[XtOrder]:
        filtered_orders: List[XtOrder] = []
        for order in orders:
            if order.order_type == order_type:
                filtered_orders.append(order)
        return filtered_orders

    @staticmethod
    def filter_order_status(orders: List[XtOrder], filtered_order_status: List[int]) -> List[XtOrder]:
        filtered_orders: List[XtOrder] = []
        for order in orders:
            if order.order_status not in filtered_order_status:
                filtered_orders.append(order)
        return filtered_orders

open_quant_app/utils/OrderFilter.py
openhe-hub/open-quant-app
from datetime import datetime, timedelta


class TimeUtils:
    @staticmethod
    def get_morning_start(timestamp: datetime = datetime.now()) -> datetime:
        return timestamp.replace(hour=9, minute=30, second=0)

    @staticmethod
    def get_morning_end(timestamp: datetime = datetime.now()) -> datetime:
        return timestamp.replace(hour=11, minute=30, second=0, microsecond=0)

    @staticmethod
    def get_afternoon_start(timestamp: datetime = datetime.now()) -> datetime:
        return timestamp.replace(hour=13, minute=0, second=0)

    @staticmethod
    def get_afternoon_end(timestamp: datetime = datetime.now()) -> datetime:
        return timestamp.replace(hour=15, minute=0, second=0)

    @staticmethod
    def before_morning_start(timestamp: datetime = datetime.now()) -> bool:
        return timestamp < TimeUtils.get_morning_start(timestamp)

    @staticmethod
    def after_afternoon_end(timestamp: datetime = datetime.now()) -> bool:
        return timestamp > TimeUtils.get_afternoon_end(timestamp)

    @staticmethod
    def get_trade_time_seg() -> [datetime]:
        morning_start = datetime.now().replace(hour=9, minute=30, second=0, microsecond=0)
        morning_end = morning_start + timedelta(hours=2)

        afternoon_start = datetime.now().replace(hour=13, minute=0, second=0, microsecond=0)
        afternoon_end = afternoon_start + timedelta(hours=2)

        trade_times = [morning_start, morning_end, afternoon_start, afternoon_end]
        return trade_times

    @staticmethod
    def judge_trade_time(timestamp: datetime) -> bool:
        return (TimeUtils.get_morning_start(timestamp) <= timestamp <= TimeUtils.get_morning_end(timestamp)) or (
                TimeUtils.get_afternoon_start(timestamp) <= timestamp <= TimeUtils.get_afternoon_end(timestamp))

    @staticmethod
    def next_trade_timestamp(timestamp: datetime, period=1) -> datetime:
        if TimeUtils.judge_trade_time(timestamp):
            return timestamp + timedelta(seconds=period)
        else:
            if TimeUtils.get_morning_end(timestamp) < timestamp < TimeUtils.get_afternoon_end(timestamp):
                return TimeUtils.get_afternoon_start(timestamp)
            elif timestamp > TimeUtils.get_afternoon_end(timestamp):
                return TimeUtils.get_morning_start(timestamp) + timedelta(days=1)
            else:
                return timestamp + timedelta(seconds=period)

open_quant_app/utils/TimeUtils.py
openhe-hub/open-quant-app
from setuptools import setup, find_packages

setup(
    name='open-quant-app',
    version='0.1.5.4',
    packages=find_packages(),
    install_requires=[
        'numpy',
        'pandas',
        'matplotlib',
        'dash',
        'loguru',
        'toml'
    ],
    author='openhe',
    author_email='625529334@qq.com',
    description='A quantitative trading framework.',
    long_description=open('README.md').read(),
    long_description_content_type="text/markdown",
    license='MIT',
    keywords='quantitative trading finance',
    url='https://github.com/openhe-hub/open-quant-app'
)

setup.py
openhe-hub/open-quant-app