table.py - master - majorow/table

table.py
440 строк19 KB
majorow
Create: test.py, Update: table.py
15 июн 2026, 06:48
15 июн 2026, 06:48552094e
from array import array
from statistics import fmean, median, stdev, linear_regression
from operator import neg, add, sub, mul, truediv
from csv import reader, writer

def isiter(data:list|tuple|set) -> bool:
  '''Проверка на итерируемость объекта, а именно имеется ли атрибуты
  __len__ и __getitem___'''
  return all(hasattr(data, x) for x in ['__len__', '__getitem__'])

def allnumeric(data:list|tuple|set) -> bool:
  '''Проверка элементов последовательности к целым числам, либо числам
  с плавающей точкой'''
  return all(map(isinstance, data, [(int, float)] * len(data)))

def isdate(text:str) -> bool:
  '''Проверка текста на принадлежность к дате в формате ISO'''
  return text.translate(str.maketrans('', '', '-: ')).isdigit()

def date2int(text:str) -> int:
  '''Конвертирование текста, содержащего дату, в целое число'''
  return int(text.translate(str.maketrans('', '', '-: ')))

def int2date(num:int) -> str:
  '''Конвертирование целого числа, содержащего дату, в текст в формате ISO'''
  if num >= 19700101 and num < 100000000:
    year = str(num // 10000)
    month = str(num // 100 % 100).zfill(2)
    day = str(num % 100).zfill(2)
    return f'{year}-{month}-{day}'

class Array():
  '''Класс, работающий с одномерным массивом одного типа, основанный
  на стандартном типе array, но может автоматически определять тип данных,
  а также, работающий с датой и текстом. Причем класс пытается преобразовать
  текстовые значения в числовые.'''

  def __init__(self, data:list|tuple|set):
    #Передается итерируемый обьект, у которого все элементы одного типа
    if not isiter(data):
      raise Exception('Надо передавать итерируемый объект')
    if all(isinstance(x, int) for x in data):
      self.data = array('i', data)
      self.dtype = 'i'
    elif all(isinstance(x, float) for x in data):
      self.data = array('f', data)
      self.dtype = 'f'
    elif all(isinstance(x, str) for x in data):
      # Пустые строки недопустимы
      if '' in data:
        raise Exception('Пустые значения недопустимы')
      # Проверяем ряд на уникальность значений
      unique = sorted(set(data))
      if len(unique) == len(data):
        if all(x.isdigit() for x in data):
          self.data = array('i', [*map(int, data)])
          self.dtype = 'u'
        elif all(isdate(x) for x in data):
          self.data = array('i', [date2int(x) for x in data])
          self.dtype = 'u'
        else:
          self.data = array('i', [*range(len(data))])
          self.dtype = 'u'
      # Проверяем на возможность преобразования в int
      elif all(x.isdigit() for x in data):
        self.data = array('i', [*map(int, data)])
        self.dtype = 'i'
      # Проверяем на возможность преобразования в float
      elif all(x.replace('.', '', 1).isdigit() for x in data):
        self.data = array('f', [*map(float, data)])
        self.dtype = 'f'
      # Дата ли
      elif all(isdate(x) for x in data):
        self.data = array('i', [date2int(x) for x in data])
        self.dtype = 'd'
      else:
        rep = {v: i for i, v in enumerate(unique)}
        self.data = array('i', [rep[x] for x in data])
        self.dtype = 'c'
        self.rep = {v: k for k, v in rep.items()}
    self.norm = [0] * 4


  def __len__(self):
    return len(self.data)

  def __str__(self) -> str:
    match self.dtype:
      case 'i' | 'f' | 'u':
        return str([*self.data])
      case 'd':
        return ' '.join([int2date(x) for x in self.date])
      case 'c':
        return ' '.join([self.rep[x] for x in self.date])

  def __getitem__(self, key):
    return self.data[key]

  def __add__(self, other) -> object:
    # Сложение вектора числового типа с числом, либо с вектором такой же длины
    if self.dtype == 'i' or self.dtype == 'f':
      if isinstance(other, (int, float)):
        return Array([*map(add, self.data, [other] * self.__len__())])
      if isiter(other) and self.__len__() == len(other) and allnumeric(other):
          return Array([*map(add, self.data, other)])

  def __sub__(self, other) -> object:
    # Вычитание из вектора числа, либо поэлементное вычитание из вектора
    if self.dtype == 'i' or self.dtype == 'f':
      if isinstance(other, (int, float)):
        return Array([*map(sub, self.data, [other] * self.__len__())])
      if isiter(other) and self.__len__() == len(other) and allnumeric(other):
          return Array([*map(sub, self.data, other)])

  def __mul__(self, other) -> object:
    # Умножение вектора на число, либо поэлементное умножение векторов
    if self.dtype == 'i' or self.dtype == 'f':
      if isinstance(other, (int, float)):
        return Array([*map(mul, self.data, [other] * self.__len__())])
      if isiter(other) and self.__len__() == len(other) and allnumeric(other):
          return Array([*map(mul, self.data, other)])

  def __truediv__(self, other) -> object:
    # Деление вектора на число, либо поэлементное деление векторов
    if self.dtype == 'i' or self.dtype == 'f':
      if isinstance(other, (int, float)):
        return Array([*map(truediv, self.data, [other] * self.__len__())])
      if isiter(other) and len(self.data) == len([*filter(None, other)]) \
        and allnumeric(other):
          return Array([*map(truediv, self.data, other)])

  def __neg__(self) -> object:
    if self.dtype == 'i' or self.dtype == 'f':
      return Array([*map(neg, self.data)])

  def isunique(self) -> bool:
    if self.dtype == 'u': return True
    return len(self.data) == len(set(self.data))

  def tolist(self) -> list:
    match self.dtype:
      case 'i' | 'f' | 'u':
        return [*self.data]
      case 'd':
        return [int2date(x) for x in self.data]
      case 'c':
        return [self.rep[x] for x in self.data]

  def max(self) -> int|float|str:
    return max(self.data)

  def min(self) -> int|float|str:
    return min(self.data)

  def mean(self) -> int|float:
    return fmean(self.data)

  def std(self) -> int|float:
    return stdev(self.data)

  def median(self) -> int|float:
    return median(self.data)

  def sum(self) -> int|float:
    if self.dtype == 'i' or self.dtype == 'f':
      return sum(self.data)

  def normalize(self) -> object:
    '''Новый объект с нормализованными числовыми данными.'''
    if self.dtype == 'i' or self.dtype == 'f':
      if self.norm[0] == 0 and self.norm[1] == 0:
        vmin, vmax = self.min(), self.max()
        new = Array([(x - vmin) / (vmax - vmin) for x in self.data])
        new.norm[0], new.norm[1] = vmin, vmax
        return new

  def standardize(self) -> object:
    '''Новый объект с стардатизированными числовыми данными.'''
    if self.dtype == 'i' or self.dtype == 'f':
      if self.norm[2] == 0 and self.norm[3] == 0:
        vmean, vstd = self.mean(), self.std()
        new = Array([(x - vmean) / vstd for x in self.data])
        new.norm[2], new.norm[3] = vmean, vstd
        return new

  def recover(self) -> object:
    '''Новый объект с восстановленными исходными данными после
    нормализации или стандартизации.'''
    if self.dtype == 'i' or self.dtype == 'f':
      if self.norm[0] != 0 or self.norm[1] != 0:
        vmin, vmax = self.norm[0], self.norm[1]
        new = Array([x * (vmax - vmin) + vmin for x in self.data])
        new.norm[0], new.norm[1] = 0, 0
        return new
      elif self.norm[2] != 0 or self.norm[3] != 0:
        new = Array([x * self.norm[3] + self.norm[2] for x in self.data])
        new.norm[2], new.norm[3] = 0, 0
        return new

  def float2int(self) -> object:
    '''Новый обьект с элементами типа int без потери'''
    if self.dtype == 'f' and all(x.is_integer() for x in self.data):
      return Array([*map(int, self.data)])

  def cov(self, other:list|tuple|set) -> float:
    '''Ковариация с другим рядом такого же размера'''
    if (self.dtype == 'i' or self.dtype == 'f') and isiter(other):
      if len(other) == self.__len__():
        a = Array(other)
        m1, m2 = self.mean(), a.mean()
        a1, a2 = self - m1, a - m2
        return (a1 * a2).sum() / (self.__len__() - 1)

  def corr(self, other:list|tuple|set) -> float:
    '''Коэфициент корреляции с другим рядом такого же размера'''
    if (self.dtype == 'i' or self.dtype == 'f') and isiter(other):
      if len(other) == self.__len__():
        a = Array(other)
        return self.cov(other) / (self.std() * a.std())

  def linregress(self, target:list|tuple|set) -> tuple[float, float]:
    '''Линейная регрессия с помощью цели такого же размера.
    Возвращает наклон и смещение'''
    if (self.dtype == 'i' or self.dtype == 'f') and isiter(target):
      if len(target) == self.__len__():
        return tuple(linear_regression(self.data, target))

  def shift(self, n:int=1) -> object:
    '''Смещение ряда на n элементов, соответственно длина ряда становится
    на n меньше'''
    if self.dtype == 'i' or self.dtype == 'f':
      return Array(self.data[:-n])

  def diff(self, n:int=1) -> object:
    '''Разница между и n элементом, соответственно длина ряда становится
    на n меньше'''
    if self.dtype == 'i' or self.dtype == 'f':
      return Array([*map(sub, self.data[n:], self.data[:-n])])

  def gradient(self) -> object:
    '''Градиент ряда'''
    if self.dtype == 'i' or self.dtype == 'f':
      reduced = self.data[1:] + self.data[-1:]
      subtract = self.data[:1] + self.data[:-1]
      denomenator = [2.] * len(self.data)
      denomenator[0], denomenator[-1] = 1., 1.
      return Array([*map(truediv, map(sub, reduced, subtract), denomenator)])

  def hist(self, bins:int=20) -> dict:
    '''Реальное распределение плотности'''
    vmin, vmax = min(self.data), max(self.data)
    step = (vmax - vmin) / bins
    res = {vmin + i * step: 0 for i in range(1, bins+1)}
    for v in self.data:
      for u in [*res.keys()][:-1]:
        if v < u:
          res[u] += 1
          break
      else:
        res[[*res.keys()][-1]] += 1
    return res

def arange(stop:int) -> Array:
    return Array([*range(stop)])

def linspace(start:float, stop:float, size:int) -> Array:
    step = (stop - start + 1) / size
    return Array([start + i * step for i in range(size)])

class Table():
  '''Класс, работающий с табличными данными (двумерными массивами).
  Первая ось - строки таблицы идентифицируются индексом, если он не
  задан, то это целые числа, начиная с 0.
  Вторая ось - колонки (столбцы, поля) идентифицируются либо именем, либо
  целым числом, начиная с 0'''

  def __init__(self, **params):
    '''Аргументы:
    - data - словарь, содержащий индексы, перечень колонок и данные
            соответсвующие данному индексу и колонке, или список или кортеж,
            представляющий двумерный массив, первая ось - строки, вторая
            ось - столбцы;
    - index - список, кортеж или множество, содержит перечень уникальных
            индексов строк;
    - columns - список, кортеж или множество, содержит перечень уникальных
            индексов колонок.'''
    match params.get('data'):
      case dict(data):
        self.rows = tuple(data.keys())
        self.head = tuple(next(iter(data.values())).keys())
        data = [[*x.values()] for x in data.values()]
        self.data = tuple(Array(x) for x in zip(*data))
      case list(data) | tuple(data):
        if not all(isiter(x) for x in data):
          raise Exception('Должен передаваться двумерный массив')
        self.data = tuple(Array(x) for x in zip(*data))
        idx = params.get('index')
        if idx and isiter(idx) and len(idx) == len(self.data[0]):
          self.rows = tuple(idx)
        else:
          self.rows = tuple(range(len(self.data[0])))
        cols = params.get('columns')
        if cols and isiter(cols) and len(cols) == len(self.data):
          self.head = cols
        else:
          self.head = tuple(range(len(self.data)))
      case _:
        self.data = None

  def __len__(self) -> int:
    '''Количество строк'''
    return len(self.rows)

  def todict(self) -> dict:
    return {x: {y: self.data[k][i] for k, y in enumerate(self.head)} for i,
                x in enumerate(self.rows)}

  def tolist(self) -> list[list]:
    return [*map(list, zip(*[x.tolist() for x in self.data]))]

  def savecsv(self, fname:str):
    with open(fname, 'w', encoding='utf-8') as f:
      wrt = writer(f)
      wrt.writerow(self.head)
      wrt.writerows(self.tolist())

  def __str__(self) -> str:
    '''Текстовое представление'''
    return str(self.todict())

  def shape(self) -> tuple[int, int]:
    '''Количество строк, количество колонок'''
    return len(self.rows), len(self.head)

  def __getitem__(self, key):
    '''Возвращает элемент (массив) по ключу'''
    match key:
      case tuple(idx):
        if isinstance(idx[0], int) and isinstance(idx[1], int):
          return self.data[idx[1]][idx[0]]
        if isinstance(idx[0], slice) and isinstance(idx[1], slice):
          row, col = self.rows[idx[0]], self.head[idx[1]]
          ret = [[self.data[self.head.index(x)][self.rows.index(y)] \
                    for x in col] for y in row]
          return Table(data=ret, index=row, columns=col)
        if isinstance(idx[0], slice) and isinstance(idx[1], int):
          row = self.rows[idx[0]]
          return Array([self.data[idx[1]][self.rows.index(x)] for x in row])
        if isinstance(idx[0], int) and isinstance(idx[1], slice):
          col = self.head[idx[1]]
          return Array([self.data[self.head.index(x)][idx[0]] for x in col])
      case int(int_idx):
        return Array([self.data[x][int_idx] for x in range(len(self.head))])
      case str(str_idx):
        if str_idx in self.rows:
          idx_int = self.rows.index(str_idx)
          return Array([self.data[x][idx_int] for x in range(len(self.head))])
        if str_idx in self.head:
          return self.data[self.head.index(str_idx)]
      case _:
        if isinstance(key, slice):
          ret = {x: {y: self.data[k][i] for k,
                     y in enumerate(self.head)} for i,
                     x in enumerate(self.rows[key])}
          return Table(data=ret)

  def groupby(self, col:str|int, agg:dict, sort:bool=True) -> object:
    dc = dict()
    acts = ['sum', 'mean', 'first', 'last', 'min', 'max']
    if col not in self.head: raise Exception(f'Отсутствует столбец {col}')
    idcol = self.head.index(col)
    for k in agg.keys():
      if k not in self.head: raise Exception(f'Отсутствует столбец {k}')
    for v in agg.values():
      if v not in acts:
        raise Exception(f'Неправильная агрегирующая команда {v}')
    idcols = [self.head.index(x) for x in agg.keys()]
    [dc.setdefault(v, []).append([self.data[x][i] for x in idcols]) for i,
                                          v in enumerate(self.data[idcol])]
    res = sorted(dc.items()) if sort else dc.items()
    resdc = dict()
    for k, v in res:
      data = [*zip(*v)]
      for i, val in enumerate(agg.values()):
        match val:
          case 'sum':
            resdc.setdefault(k, []).append(sum(data[i]))
          case 'mean':
            resdc.setdefault(k, []).append(fmean(data[i]))
          case 'first':
            resdc.setdefault(k, []).append(data[i][0])
          case 'last':
            resdc.setdefault(k, []).append(data[i][-1])
          case 'min':
            resdc.setdefault(k, []).append(min(data[i]))
          case 'max':
            resdc.setdefault(k, []).append(max(data[i]))
    return resdc

  def zigzag(self, mincol:str, maxcol:str, limit:float=0.05) -> dict:
    if mincol not in self.head: return None
    if maxcol not in self.head: return None
    high, prev, ret = self.data[self.head.index(maxcol)], 0, dict()
    size, low = len(high), self.data[self.head.index(mincol)]
    for i in range(1, size - 1):
      vmax = high[i]
      if high[i-1] < vmax > high[i+1] and i - prev > 5:
        min_range = low[prev+1:i]
        vmin = min(min_range)
        if (high[prev]-vmin)/vmin>limit and (vmax-vmin)/vmin>limit:
          imin = min_range.index(vmin) + prev + 1
          ret[int2date(self.rows[prev])] = {'size': imin - prev,
                                        'open': high[prev], 'close': vmin}
          ret[int2date(self.rows[imin])] = {'size': i - imin,
                                            'open': vmin, 'close': vmax}
          prev = i
    min_range = low[prev+1:size]
    vmin = min(min_range)
    imin = min_range.index(vmin) + prev + 1
    ret[int2date(self.rows[prev])] = {'size': imin - prev,
                                      'open': high[prev], 'close': vmin}
    ret[int2date(self.rows[imin])] = {'size': size - imin - 1,
                                      'open': vmin, 'close': high[-1]}
    return ret

def readcsv(fname:str) -> Table:
  with open(fname) as f:
    rdr = reader(f)
    head = next(rdr)
    data = tuple(zip(*rdr))
  idx = [i for i, v in enumerate(data) if not '' in v]
  head = [head[i] for i in idx]
  data = [Array(data[i]) for i in idx]
  idx = [i for i, v in enumerate(data) if v.dtype == 'u']
  index = data[idx[0]]
  idx = [i for i, v in enumerate(data) if v.dtype != 'u']
  head = [head[i] for i in idx]
  data = [Array(data[i]) for i in idx]
  return Table(data={x: {y: data[k][i] for k, y in enumerate(head)} for i,
                     x in enumerate(index)})