		Metadata-Version: 2.4
		Name: vxutils
		Version: 20251226
		Version: 20251230
		Summary: A toolbox for vxquant
		@@ -5,0 +5,0 @@ Author-email: libao <libao@vxquant.com>

+1

-1

pyproject.toml

		[project]
		name = "vxutils"
		version = "20251226"
		version = "20251230"
		description = "A toolbox for vxquant"
		@@ -5,0 +5,0 @@ readme = "README.md"

+16

-6

src/vxutils/datamodel/core.py

		"""基础模型"""

		import datetime
		import enum
		from typing import Any, Dict
		@@ -13,3 +14,3 @@ from pydantic import (
		)
		from vxutils.convertors import to_datetime, VXJSONEncoder
		from vxutils.convertors import to_datetime, VXJSONEncoder, to_timestr, to_json

		@@ -25,3 +26,12 @@ try:
		class VXDataModel(BaseModel):
		model_config = ConfigDict(arbitrary_types_allowed=True)
		model_config = ConfigDict(
		arbitrary_types_allowed=True,
		validate_assignment=True,
		json_encoders={
		datetime.datetime: to_timestr,
		datetime.date: to_timestr,
		datetime.time: to_timestr,
		enum.Enum: lambda v: v.value,
		},
		)

		@@ -60,6 +70,6 @@ updated_dt: Annotated[datetime.datetime, PlainValidator(to_datetime)] = Field(
		def __str__(self) -> str:
		return self.model_dump_json(indent=4, warnings="none")
		return to_json(self)

		def __repr__(self) -> str:
		return self.model_dump_json(indent=4, warnings="none")
		return to_json(self)

		@@ -86,6 +96,6 @@ @field_validator("updated_dt", "created_dt", mode="plain")
		tick = vxTick(symbol="123")
		pprint(tick.__pydantic_core_schema__)
		# pprint(tick.__pydantic_core_schema__)
		tick.updated_dt = "2021-01-01 00:00:00"
		tick.trigger_dt = "2021-01-01 00:00:00"
		pprint(tick.model_fields)
		# pprint(tick.__class__.model_fields)

		@@ -92,0 +102,0 @@ print(tick)

+0

-0

src/vxutils/executor.py

@@ -0,0 +0,0 @@ import os

-77

src/vxutils/datamodel/polars_orm.py

		from typing import Type, Dict, Any
		import polars as pl
		from enum import Enum
		from datetime import datetime, date, time, timedelta
		from vxutils.datamodel import VXDataModel

		__columns_mapping__: Dict[Any, pl.DataType] = {
		int: pl.Int64,
		float: pl.Float64,
		bool: pl.Boolean,
		bytes: pl.Binary,
		str: pl.Utf8,
		Enum: pl.Utf8,
		datetime: pl.Datetime,
		date: pl.Date,
		time: pl.Time,
		timedelta: pl.Float64,
		}


		class PolarsORM:
		def __init__(self, model_cls: Type[VXDataModel], keys: list[str] = None):
		self._model_cls = model_cls
		self._keys = keys or []
		self._data: pl.DataFrame = pl.DataFrame(
		data=[{"name": None for name in self._model_cls.model_fields.keys()}],
		schema={
		name: __columns_mapping__.get(field.annotation, pl.Utf8)
		for name, field in self._model_cls.model_fields.items()
		},
		).clear()

		@property
		def data(self) -> pl.DataFrame:
		return self._data

		def save(self, *data: VXDataModel) -> None:
		if not all(isinstance(item, self._model_cls) for item in data):
		raise ValueError(f"Invalid data type: {type(data)}")

		if self._keys:
		self._data = self._data.filter(
		pl.any(
		pl.all(pl.col(key) != item[key] for key in self._keys)
		for item in data
		).not_()
		)
		self._data = pl.concat(
		[
		self._data,
		pl.DataFrame([item.model_dump() for item in data]).select(
		pl.col(self._data.columns)
		),
		]
		)


		if __name__ == "__main__":
		df = pl.DataFrame(
		data=[
		{"id": 1, "name": "a"},
		{"id": 2, "name": "b"},
		],
		schema={
		"id": pl.Int64,
		"name": pl.Utf8,
		},
		)
		print(df)

		class A(VXDataModel):
		id: int
		name: str

		porm = PolarsORM(A, keys=["id"])
		porm.save(A(id=1, name="c"))
		print(porm.data)

vxutils - npm Package Compare versions

Worsened metrics