Coverage for core/models/stock.py: 100.00%

1""" 

2股票数据模型 

3""" 

4 

5from datetime import datetime 

6from enum import Enum 

7from typing import Optional 

8 

9from pydantic import BaseModel, Field 

10 

11 

12class TradeSession(str, Enum): 

13 """交易时段枚举""" 

14 

15 INTRADAY = "Intraday" # 盘中交易 

16 PRE = "Pre" # 盘前交易 

17 POST = "Post" # 盘后交易 

18 OVERNIGHT = "Overnight" # 夜盘交易 

19 

20 

21class StockData(BaseModel): 

22 """股票数据模型""" 

23 

24 code: str = Field(..., description="股票代码", example="YINN.US") 

25 open: float = Field( 

26 ..., 

27 description="开盘价", 

28 ge=0, 

29 ) 

30 high: float = Field( 

31 ..., 

32 description="最高价", 

33 ge=0, 

34 ) 

35 low: float = Field( 

36 ..., 

37 description="最低价", 

38 ge=0, 

39 ) 

40 close: float = Field( 

41 ..., 

42 description="收盘价", 

43 ge=0, 

44 ) 

45 volume: int = Field(..., description="成交量", ge=0) 

46 turnover: float = Field( 

47 ..., 

48 description="成交额", 

49 ge=0, 

50 ) 

51 timestamp: int = Field(..., description="时间戳(UNIX时间戳)", example=1704067200) 

52 trade_session: TradeSession = Field(..., description="交易时段") 

53 

54 class Config: 

55 json_encoders = {datetime: lambda v: int(v.timestamp())} 

56 

57 

58class StockDataCreate(BaseModel): 

59 """创建股票数据模型""" 

60 

61 code: str = Field(..., description="股票代码") 

62 open: float = Field( 

63 ..., 

64 description="开盘价", 

65 ge=0, 

66 ) 

67 high: float = Field( 

68 ..., 

69 description="最高价", 

70 ge=0, 

71 ) 

72 low: float = Field( 

73 ..., 

74 description="最低价", 

75 ge=0, 

76 ) 

77 close: float = Field( 

78 ..., 

79 description="收盘价", 

80 ge=0, 

81 ) 

82 volume: int = Field(..., description="成交量", ge=0) 

83 turnover: float = Field( 

84 ..., 

85 description="成交额", 

86 ge=0, 

87 ) 

88 timestamp: int = Field(..., description="时间戳(UNIX时间戳)") 

89 trade_session: TradeSession = Field(..., description="交易时段") 

90 

91 

92class StockDataFilter(BaseModel): 

93 """股票数据筛选模型""" 

94 

95 code: Optional[str] = Field(None, description="股票代码筛选") 

96 start_timestamp: Optional[int] = Field(None, description="开始时间戳") 

97 end_timestamp: Optional[int] = Field(None, description="结束时间戳") 

98 trade_session: Optional[TradeSession] = Field(None, description="交易时段筛选") 

99 sort_by: str = Field("timestamp", description="排序字段") 

100 sort_order: str = Field("asc", description="排序方式: asc/desc") 

101 page: int = Field(1, description="页码", ge=1) 

102 page_size: int = Field(120, description="每页数量", ge=1, le=1440) 

103 timezone: str = Field("Asia/Shanghai", description="时区") 

104 

105 

106class StockDataResponse(BaseModel): 

107 """股票数据响应模型""" 

108 

109 data: list[StockData] = Field(..., description="股票数据列表") 

110 total: int = Field(..., description="总数量") 

111 page: int = Field(..., description="当前页码") 

112 page_size: int = Field(..., description="每页数量") 

113 total_pages: int = Field(..., description="总页数") 

114 

115 

116class StockCodeList(BaseModel): 

117 """股票代码列表模型""" 

118 

119 codes: list[str] = Field(..., description="股票代码列表")