创建股票单位或股票集合的类

Question

我有一个单一类对象'AStock‘的数据集，如下所示：

@dataclass
class AStock:
    
    code_: Union[str, int]
    force_update_: bool = False
    ticker_: str = field(init=False)
    symbol_: str = field(init=False)
    name_: str = field(init=False)
    sector_: str = field(init=False)
    subsector_: str = field(init=False)
    
    @exception(logger)
    def __post_init__(self) -> None:
        try:
            self.code_ = str(self.code_).zfill(4)
            self.ticker_ = indexer(code=self.code_, arg='ticker')
            self.symbol_ = indexer(code=self.code_, arg='symbol')
            self.name_ = indexer(code=self.code_, arg='full_name')
            self.sector_ = indexer(code=self.code_, arg='sector')
            self.subsector_ = indexer(code=self.code_, arg='subsector')

        except NoStockFound:
            raise
        except Exception as e:
            raise StockException(stock=self) from e

我想要实现的是创建一个类对象，它将初始化单个股票对象或股票对象的集合。所以我做了一个‘股票’和‘股票’对象。

这里的‘'Stocks’初始化一组'AStock‘对象。

@dataclass
class Stocks:
    
    code_: List[str]
    force_update_: bool = False
    ticker_: str = field(init=False)
    symbol_: str = field(init=False)
    name_: str = field(init=False)
    sector_: str = field(init=False)
    subsector_: str = field(init=False)
    
    @exception(logger)
    def __post_init__(self) -> None:
        self.ticker_ = [indexer(code, arg='ticker') for code in self.code_]
        self.symbol_ = [indexer(code, arg='symbol') for code in self.code_]
        self.name_ = [indexer(code, arg='full_name') for code in self.code_]
        self.sector_ = [indexer(code, arg='sector') for code in self.code_]
        self.subsector_ = [indexer(code, arg='subsector') for code in self.code_]
    
    def __repr__(self) -> str:
        return ', '.join([f'{i[0]}:{i[1]}' for i in list(zip(self.code_, self.ticker_))])
    
    def __str__(self) -> str:
        return ', '.join([f'{i[0]}:{i[1]}' for i in list(zip(self.code_, self.ticker_))])

这里的“股票”对象是处理“AStock”或“Stock”对象的创建。

@dataclass
class Stock:
    code: Union[str, int, List[str]]
    force_update: bool = False
    
    @exception(logger)
    def __new__(cls, code, force_update: bool = False):
        if type(code) in [str, int]:
            return AStock(code, force_update)
        
        elif type(code) in [list]:
            return Stocks(code, force_update)
        
        else:
            raise ValueError('Not a valid stock code')

例如:如果Stock(['1203', '1232'])将创建Stocks对象，Stock('1203')将创建AStock对象。只要提供参数作为列表或字符串，它就可以检测是为单个股票创建AStock，还是为多个股票创建Stocks。

但是，这是使用__new__返回AStock对象或Stocks对象的“黑客”方式。有什么更优雅的方法吗？

编辑:2021年12月9日：

从下面的评论中，我想我应该展示更多我的Stocks代码。首先，是的，Stocks和AStock中有更多的方法。基本上，Stocks.any_methods将调用相同的AStock.any_methods，但是它将运行它来收集codes并输出Dict[str, output_data_type_from_AStock]。

对于上下文，这里是来自Stocks的模式代码。

# %%
@dataclass
class Stocks:
    
    code_: List[Union[str, int]]
    force_update_: bool = False
    ticker_: Tuple[str] = field(init=False)
    symbol_: Tuple[str] = field(init=False)
    name_: Tuple[str] = field(init=False)
    sector_: Tuple[str] = field(init=False)
    subsector_: Tuple[str] = field(init=False)
    
    @exception(logger)
    def __post_init__(self) -> None:
        self.ticker_ = tuple([indexer(code, arg='ticker') for code in self.code_])
        self.symbol_ = tuple([indexer(code, arg='symbol') for code in self.code_])
        self.name_ = tuple([indexer(code, arg='full_name') for code in self.code_])
        self.sector_ = tuple([indexer(code, arg='sector') for code in self.code_])
        self.subsector_ = tuple([indexer(code, arg='subsector') for code in self.code_])
    
    def __str__(self) -> str:
        return ', '.join([f'{i[0]}:{i[1]}' for i in list(zip(self.code_, self.ticker_))])
    
    def _output_to_dict(self, time_sleep: bool, *args, **kwargs) -> Union[Dict[str, pd.DataFrame], Dict[str, float]]:
        '''
        method takes any method in this class and calls same method of AStock class for each code in codes.
        
        time_sleep = True required for method that utilizes investpy which scrapes from investing.com to avoid
        ip being blocked.
        
        '''
        results = {}
        method_name = inspect.stack()[1][3]
        for code in self.code_:
            if args or kwargs:
                results[code] = getattr(
                    AStock(code, self.force_update_),
                    method_name
                )(*args, **kwargs)
            else:
                results[code] = getattr(
                    AStock(code, self.force_update_),
                    method_name
                )
            if time_sleep:
                time.sleep(5)
        return results
    
    '''
    ------------------------------------------------------
    GENERAL INFO
    ------------------------------------------------------
    '''
    
    @property
    @exception(logger)
    def info_(self) -> Dict[str, pd.DataFrame]:
        return self._output_to_dict(time_sleep = True)

    '''
    ------------------------------------------------------
    FINANCE METRICS
    ------------------------------------------------------
    '''
    
    @property
    @exception(logger)
    def ttm_financials_(self) -> Dict[str, pd.DataFrame]:
        return self._output_to_dict(time_sleep = False)

这就是为什么我需要另一个类的Stock，它将调用AStock来表示单个股票代码，或者调用Stocks来调用股票代码列表：

@dataclass
class Stock:
    
    code: Union[str, int, List[str]]
    force_update: bool = False
    
    @exception(logger)
    def __new__(
        cls,
        code: Union[str, int, List[str]],
        force_update: bool = False
    ) -> Union[AStock, Stocks]:
        
        if type(code) in [str, int]:
            return AStock(code, force_update)
        
        elif type(code) in [list]:
            return Stocks(code, force_update)
        
        else:
            raise ValueError('Not a valid stock code')

感谢@ShapeOfMatter，下面是我对Stocks所做的事情：

class Stocks:
    
    def __init__(self, *codes, force_update_: bool = False):
        self._stocks = [
            AStock(code, force_update_=force_update_) for code in codes
        ]

    def __iter__(self):
        '''
        this makes Stocks an iterable hence could you [s for s in self]
        '''
        yield from self._stocks
    
    def __str__(self) -> str:
        return ', '.join([f'{s.code_}:{s.ticker_}' for s in self])
    
    def _dict_apply(self, function: Callable, sleep: bool) -> Any:
        return {f'{stock.code_}': function(stock) for stock in self._stocks if time.sleep(5 * sleep) is None}
    
    def __getattr__(self, attr) -> Any:
        candidate = lambda x: getattr(x, attr, None)
        return partial(self._dict_apply, function=candidate)(sleep=True)
    
```

Answer 1

只要有可能，就喜欢普通的命名。如果一个类被称为Stock，这是一个非常合理的名称:几乎所有的英语使用者都会正确地假设它指的是一个单一的股票。这就是语言的工作方式，你应该利用语言约定来发挥你的优势。没有必要通过给类起一个像AStock这样的名字来使事情变得过于复杂。把一篇不确定的文章粘在名字上一点用处都没有，它只会产生一个非惯用的名字。

不要仅仅因为需要一个实例集合就创建一个类。Python已经有了方便的集合类:列表、元组、数据集等等。如果您有一堆Stock实例，并且需要将它们作为集合进行操作，那么只需使用普通的Python集合--在您的示例中，可能需要使用列表。只有当集合本身需要有意义的属性或行为时，才需要创建一个特殊的类来表示这样的集合。至少在您向我们展示的代码中，Stocks类没有这些:它纯粹是底层Stock类(即您称为AStock的类)的衍生物。底线:只要使用一个列表，除非你的计划与我们看到的不同。

如果一个简单的函数可以的话，不要创建一个类。在您当前的Stock代码中，没有任何类可以像类一样。它只是一种if-else机制，用于创建一个或多个事物集合。类是属性和行为。你只有行为。用一个函数来处理类似的事情。

Answer 2

关于编辑的

：

您似乎已经专注于解决一个简单问题的糟糕解决方案(您的Stock类)(以一种流畅的方式处理单个股票和股票组)。所有的FMc点仍然很好。

按照您现在编写的方式，MyPy应该认为Stock(87)的类型是什么？API的使用者在这些对象上调用isinstance需要多长时间？不要找到解决这些问题的方法；放弃你现有的解决方案。Stock应该只是一个函数。

您的Stocks类也有问题。有些事情我们希望能够用一个集合来完成，比如迭代成员。每当调用一个方法时，重新创建所有的AStocks也是效率低下的。您确定不能让用户对AStock对象列表进行理解吗？

如果你做不到，可以考虑这样的事情：

StockID = Union[str, int]

class Stocks:

    def __init__(self, codes: Iterable[StockID], force_update: bool = False):
        self._stocks = [AStock(code, force_update=force_update)
                        for code in codes]

    def __iter_(self):
        yield from self._stocks
    
    def __str__(self) -> str:
        return ', '.join([f'{s.code}:{s.ticker}' for s in self])
    
    def dict_apply(self,
                   function: Callable[[AStock, Any...] pd.DataFrame],  # ??
                   *args,
                   **kwargs,
                   time_sleep: bool = False)
                   -> Union[Dict[str, pd.DataFrame], Dict[str, float]]:
        '''time_sleep = True required for method that utilizes investpy which scrapes from investing.com to avoid ip being blocked.'''
        return {function(stock, *args, **kwargs)  # it's fine if args/kwargs are empty.
                for stock in self
                if (time_sleep and time.sleep(5)  # IDK if this works
                   ) or True}

    def __getattr__(self, attr):
        # If we're careful, this is probably less fragile than inspecting the call stack. Still kinda sketchy!
        candidate = getAttr(AStock, attr, None)
        if callable(candidate):
            return functools.partial(self.dict_apply, function=candidate)
        else:
            raise AttributeError, attr

以上内容包括我的一些猜测，当然还有其他的选择。一种选择是实际从list继承；这可能很有趣，但也有它自己的缺陷.