基于Python实现的量化交易系统源码+项目说明.zip

from __future__ import annotations import os import sys from math import ceil from os import walk from types import FunctionType import json import pandas as pd import numpy as np import datetime from vector import Vector class Date: """日期""" def __init__(self, init_time: datetime.datetime = None): if init_time is None: self._date = datetime.datetime.today() else: self._date = init_time def add_days(self, day): self._date += datetime.timedelta(day) def add_month(self, month: int): new = self.month + month if new == 0: ryear = -1 rmon = 12 else: ryear = new // 12 rmon = new % 12 day = self._date.day new_date = datetime.datetime(self.year + ryear, rmon, 1, self.hour, self.minute, self.second) new_date += datetime.timedelta(day - 1) while new_date.month != rmon: new_date -= datetime.timedelta(1) self._date = new_date def add_year(self, year: int): new_date = datetime.datetime(self.year + year, self.month, 1, self.hour, self.minute, self.second) new_date += datetime.timedelta(self.day - 1) self._date = new_date def add_hour(self, hour: int): self._date += datetime.timedelta(hours=hour) def add_minute(self, minute: int): self._date += datetime.timedelta(minutes=minute) def add_second(self, second: int): self._date += datetime.timedelta(seconds=second) def set_date(self, year: int, month: int, day: int): self._date = datetime.datetime(year, month, day) def set_time(self, time: datetime.datetime): self._date = time @property def year(self): return self._date.year @property def month(self): return self._date.month @property def day(self): return self._date.day @property def date(self): return self.year, self.month, self.day @property def hour(self): return self._date.hour @property def minute(self): return self._date.minute @property def second(self): return self._date.second @property def weekday(self): return self._date.weekday() def get_inter(self) -> datetime.datetime: return self._date def __str__(self): return self._date.isoformat() class KLine: """k线""" def __init__(self, date: datetime.datetime, src: dict[str, int | float]): self.date = date # 日期 self.open = src['open'] # 开盘价 self.high = src['high'] # 最高价 self.low = src['low'] # 最低价 self.close = src['close'] # 收盘价 self.volume = src['volume'] # 成交量 self.previous: KLine | None = None # 上一日k self.next: KLine | None = None # 下一日k self.vec: Vector | None = None # 向量值 @staticmethod def increasement(before: int | float, after: int | float): """计算涨幅""" if before == 0: return 0 res = after / before - 1 if before < 0: return -res return res @property def increase_day(self): """日内涨幅""" return self.increasement(self.open, self.close) @property def last_close(self): """昨日收盘价""" return self.previous.close if self.previous else self.open @property def increase(self): """涨幅""" return self.increasement(self.last_close, self.close) @property def high_offset(self): """高点偏移""" return self.increasement(self.open, self.high) @property def low_offset(self): """低点偏移""" return self.increasement(self.open, self.low) @property def daily_limit(self): """一字涨停""" return self.increase >= .04 and self.close == self.high == self.open @property def limit_down(self): """一字跌停""" return self.increase <= -.04 and self.close == self.low == self.open def ma(self, cycle: int, prop: str): """移动均线""" total = 0 ptr = self i = 0 while i < cycle: if ptr is None: break total += ptr.__getattribute__(prop) ptr = ptr.previous i += 1 return total / i def ndup(self, n: int): """连续n日上涨""" ptr = self while n > 0: if ptr is None: return False if ptr.increase <= 0: return False ptr = ptr.previous n -= 1 return True def nddown(self, n: int): """连续n日下跌""" ptr = self while n > 0: if ptr is None: return False if ptr.increase > 0: return False ptr = ptr.previous n -= 1 return True def get_history_value(self, span: int, prop: str): """获取历史属性值""" ptr = self while span: if ptr.previous is None: break ptr = ptr.previous span -= 1 return ptr.__getattribute__(prop) def interval_max(self, span: int, prop: str): """获取属性的区间最大值""" ptr = self res = ptr.__getattribute__(prop) while span: if ptr.previous is None: break ptr = ptr.previous if ptr.__getattribute__(prop) > res: res = ptr.__getattribute__(prop) span -= 1 return res def interval_min(self, span: int, prop: str): """获取属性的区间最小值""" ptr = self res = ptr.__getattribute__(prop) while span: if ptr.previous is None: break ptr = ptr.previous if ptr.__getattribute__(prop) < res: res = ptr.__getattribute__(prop) span -= 1 return res def rsi(self, m1: int = 6, m2: int = 12, m3: int = 24): """返回股票的rsi指标值""" if not m1 < m2 < m3: raise ValueError('参数必须从小到大排列！') res = [] for c in (m1, m2, m3): ptr = self a = 0 b = 0 n = 0 while ptr.previous is not None and n < c: ptr = ptr.previous n += 1 while n: i = ptr.close - ptr.previous.close if ptr.previous else ptr.close - ptr.open if i >= 0: a = ((c - 1) * a + i) / c else: b = ((c - 1) * b + i * -1) / c n -= 1 ptr = ptr.next res.append(0 if a == 0 else a / (a + b)) return res def get_datetime(self): return pd.to_datetime(self.date) def insert(self, kline: KLine): """向后插入k线""" kline.next = self.next self.next = kline kline.previous = self if kline.next is not None: kline.next.previous = kline def insert_ahead(self, kline: KLine): """向前插入k线""" kline.previous = self.previous self.previous = kline kline.next = self if kline.previous is not None: kline.previous.next = kline def __str__(self): res = ( "Open: \t{}\n".format(self.open), "High: \t{}\n".format(self.high), "Low: \t{}\n".format(self.low), "Close: \t{}\n".format(self.close), "Volume: \t{}\n".format(self.volume), "Increase: \t{:.4f}\n".format(self.increase), "Increase Day: \t{:.4f}\n".format(self.increase_day), "High Offset: \t{:.4f}\n".format(self.high_offset), "Low Offset: \t{:.4f}\n".format(sel