Python typing 模块，FrozenSet() 实例源码

def down(self, visited: FrozenSet[Any]=None) -> Set[Any]:
        """ Returns the set of reachable nodes by going down on this node """
        if visited is None:
            visited = frozenset()

        if self in visited:
            return {self} if self.symbol not in OPERATORS else set()

        visited |= {self}
        if self.symbol == '|':
            return self.left.down(visited) | self.right.down(visited)
        elif self.symbol == '.':
            return self.left.down(visited)
        elif self.symbol == '*' or self.symbol == '?':
            return self.left.down(visited) | self.right.up(visited)
        elif self.symbol == EPSILON:
            return self.right.up(visited)
        return {self}

def up(self, visited: FrozenSet[Any]=None) -> Set[Any]:
        """ Returns the set of reachable nodes by going up on this node """
        if visited is None:
            visited = frozenset()

        if self.symbol == '|':
            # skip the whole right sub tree
            node = self.right
            while node.symbol == '.' or node.symbol == '|':
                node = node.right
            return node.right.up(visited)
        elif self.symbol == '.':
            return self.right.down(visited)
        elif self.symbol == '*':
            return self.left.down(visited) | self.right.up(visited)
        elif self.symbol == '?':
            return self.right.up(visited)
        else:  # self.symbol == END:
            return {self}

def _are_undistinguishable(
            self, state_a: str, state_b: str,
            undistinguishable: Set[FrozenSet[str]]) -> bool:
        """
            State a and b are distinguishable if they go to distinguishable
            states for some input symbol.
        """
        for symbol in self._alphabet:
            transition_a = \
                list(self._transitions.get((state_a, symbol), {""}))[0]
            transition_b = \
                list(self._transitions.get((state_b, symbol), {""}))[0]
            if transition_a != transition_b and \
                    frozenset((transition_a, transition_b)) not in \
                    undistinguishable:
                return False
        return True

def __init__(self, discovered_files: etl.file_sets.TableFileSet) -> None:
        # Basic properties to locate files describing the relation
        self._fileset = discovered_files
        if discovered_files.scheme == "s3":
            self.bucket_name = discovered_files.netloc
            self.prefix = discovered_files.path
        else:
            self.bucket_name = None
            self.prefix = None
        # Note the subtle difference to TableFileSet--here the manifest_file_name is always present since it's computed
        self.manifest_file_name = os.path.join(discovered_files.path or "", "data", self.source_path_name + ".manifest")
        # Lazy-loading of table design and query statement and any derived information from the table design
        self._table_design = None  # type: Optional[Dict[str, Any]]
        self._query_stmt = None  # type: Optional[str]
        self._dependencies = None  # type: Optional[FrozenSet[str]]
        self._is_required = None  # type: Union[None, bool]

def merge_equivalent(self) -> None:
        """ Merges equivalent states """
        if not self.is_deterministic():
            raise RuntimeError("Automata is non-deterministic")

        # pairs of undistinguishable states
        undistinguishable = set()  # type: Set[FrozenSet[str]]

        # initially, you can't distinguish final and non-final states
        for pair in combinations(self._states - self._final_states, 2):
            undistinguishable.add(frozenset(pair))
        for pair in combinations(self._final_states, 2):
            undistinguishable.add(frozenset(pair))

        # find new distinguishable states
        while True:
            new_distinguishable_found = False
            undistinguishable_copy = undistinguishable.copy()
            for state_a, state_b in undistinguishable_copy:
                if not self._are_undistinguishable(
                        state_a, state_b, undistinguishable_copy):
                    undistinguishable.remove(frozenset((state_a, state_b)))
                    new_distinguishable_found = True
            if not new_distinguishable_found:
                break

        for state_a, state_b in undistinguishable:
            self._merge_states(state_a, state_b)

def pytest_assertrepr_compare(op: str, left, right) -> Optional[Sequence[str]]:
    # set of entities
    if op == '==' and isinstance(left, (set, frozenset)) and \
       isinstance(right, (set, frozenset)) and \
       all(isinstance(v, Entity) for v in left) and \
       all(isinstance(v, Entity) for v in right):
        def repr_ids(ids: AbstractSet[EntityId]) -> str:
            sorted_ids = sorted(
                ids,
                key=lambda i: (
                    i[0],
                    # Since EntityIds usually consist of one letter followed
                    # by digits, order them numerically.  If it's not in
                    # that format they should be sorted in the other bucket.
                    (0, int(i[1:])) if i[1:].isdigit() else (1, i[1:])
                )
            )
            return '{' + ', '.join(sorted_ids) + '}'
        left = cast(Union[Set[Entity], FrozenSet[Entity]], left)
        right = cast(Union[Set[Entity], FrozenSet[Entity]], right)
        left_ids = {e.id for e in left}
        right_ids = {e.id for e in right}
        return [
            '{} == {}'.format(repr_ids(left_ids), repr_ids(right_ids)),
            'Extra entities in the left set:',
            repr_ids(left_ids - right_ids),
            'Extra entities in the right set:',
            repr_ids(right_ids - left_ids),
        ]
    return None

def check_frozenset(a: typing.FrozenSet[int]) -> typing.FrozenSet[int]:
    return a

def fields(self) -> FrozenSet[str]:
        if self._fields is None:
            self._fields = frozenset(field['SystemName'] for field in self.table)
        return self._fields

def vocabulary(self) -> FrozenSet[str]:
        return self.properties.vocabulary

def _get_vocabulary(self) -> FrozenSet[str]:
        source = ''.join(text_map)
        source_special = ' '
        target = ''.join(np.arange(0, self._digits).astype(np.str))
        return frozenset(source + source_special + target)

def user_roles(self) -> typing.FrozenSet[str]:
        return frozenset(self._user_roles)

def user_roles_indexable(self) -> typing.FrozenSet[str]:
        return frozenset(self._user_roles_indexable)

def user_caps(self) -> typing.Mapping[str, typing.FrozenSet[str]]:
        return self._user_caps

def variables(self) -> FrozenSet[str]:
        """The names of the variables the constraint depends upon.

        Used by matchers to decide when a constraint can be evaluated (which is when all
        the dependency variables have been assigned a value). If the set is empty, the constraint will
        only be evaluated once the whole match is complete.
        """
        return frozenset()

def dependencies(self) -> FrozenSet[str]:
        if self._dependencies is None:
            self._dependencies = frozenset(self.table_design.get("depends_on", []))
        return self._dependencies

def _possible_investments(self, market_state: MarketState) -> FrozenSet[str]:
        '''
        Returns a set of all coins that the strategy might invest in, excluding
        `self.fiat`.
        '''
        return market_state.available_coins() - {self.fiat}

def _held_coins(self) -> FrozenSet[str]:
        return frozenset(
            coin for (coin, balance)
            in self.balances.items()
            if balance > 0
        )

def available_markets(self) -> FrozenSet[str]:
        """Return a frozenset containing all available markets, e.g. 'BTC_ETH'.
        """
        return frozenset(
            filter(
                lambda market: market.startswith(self.fiat),
                self.chart_data.keys(),
            )
        )

def available_coins(self) -> FrozenSet[str]:
        markets = self.available_markets()  # All of these start with fiat
        return frozenset(split_currency_pair(m)[1] for m in markets) | {self.fiat}

def held_coins_with_chart_data(self) -> FrozenSet[str]:
        return self._held_coins() & self.available_coins()

def __init__(self,
                 source_lang: str, target_lang: str,
                 key: Any=None,
                 vocabulary: FrozenSet[str]=None,
                 observations: int=None,
                 validate: bool=False,
                 name: str='unamed',
                 tqdm: bool=True,
                 **kwargs) -> None:
        # save basic properties
        self._show_tqdm = tqdm

        # compute properties
        computed_properties = self._fetch_corpus_properties(
            name, key, observations=observations
        )
        # use computed vocabulary if not provided
        if vocabulary is None:
            vocabulary = computed_properties.vocabulary

        # make properties public
        self.properties = CorpusProperties(
            vocabulary=vocabulary,
            histogram=computed_properties.histogram
        )

        # validate properties
        # http://unicode-search.net/unicode-namesearch.pl
        if '^' in vocabulary:
            raise ValueError('a eos char (^) was found in the vocabulary')
        if '?' in vocabulary:
            raise ValueError('a null char (?) was found in the vocabulary')
        if '' in vocabulary:
            raise ValueError('an invalid char () was found in the vocabulary')

        # create encoding schema
        self._setup_encoding()

        # set language properties
        self.source_lang = source_lang
        self.target_lang = target_lang

        # validate corpus properties
        if validate:
            self._validate_corpus_properties(name)

        # setup tensorflow pipeline
        super().__init__(histogram=self.properties.histogram,
                         dtype=self._encode_dtype,
                         name=name,
                         tqdm=tqdm,
                         **kwargs)

def word_vector_sum(
        documents: Iterable[str],
        vector_dict: Dict[str, np.array],
        *,
        separator: Optional[FrozenSet] = None,
        lowercase: bool = True,
        stop_words: Optional[Set] = None
):
    """
    Calculate the word vector sum for each document in an iterable.

    Args:
        documents: An iterable of strings representing text
            documents.

        vector_dict(dict): A dictionary of words and their
            corresponding vectors.

        separator(frozenset): A frozenset of Unicode code points
            to use as separators. Defaults to
            ``cypunct.unicode_classes.COMMON_SEPARATORS``.

        lowercase(bool): If True, lowercase each document before
            processing it.

        stop_words(set): A set of words whose vectors should not
            be included in the sum.

    Returns:
        (np.array): Return a numpy array containing a word
            vector for each document, where each document's
            word vector is the sum of its word vectors.

    """
    # Dimension of the vectors we're using.
    dimension = len(next(vector_dict.values().__iter__()))
    # If every word in a document is either a stop word or
    # not in the dictionary...
    # then the list comprehension inside np.average will return
    # []. For that case, we will insert an array of zeros.
    is_empty = [np.zeros(dimension)]
    if lowercase:
        documents = (doc.lower() for doc in documents)
    if stop_words:
        return np.asarray([
            np.average([
                vector_dict[x]
                for x in cypunct.split(doc, separator)
                if x in vector_dict and
                x not in stop_words
            ] or is_empty, axis=0)
            for doc in documents
        ])
    return np.asarray([
        np.average([
            vector_dict[x]
            for x in cypunct.split(doc, separator)
            if x in vector_dict
        ] or is_empty, axis=0)
        for doc in documents
    ])

def propose_trades_for_partial_rebalancing(
        self,
        market_state: MarketState,
        coins_to_rebalance: FrozenSet[str],
    ) -> List[AbstractTrade]:
        """TODO: Trade directly from X to Y without going through fiat.
        """
        ideal_fiat_value_per_coin = self._ideal_fiat_value_per_coin(market_state)

        est_values = market_state.estimate_values(market_state.balances, self.fiat)

        # 1) Fan in to fiat, selling excess value in coins we want to rebalance
        trades_to_fiat = []
        for sell_coin in sorted(coins_to_rebalance):
            if sell_coin == self.fiat:
                continue
            value = est_values.get(sell_coin, 0)
            delta = value - ideal_fiat_value_per_coin
            if delta > 0:
                trades_to_fiat.append(
                    AbstractTrade(sell_coin, self.fiat, self.fiat, delta),
                )

        # 2) Simulate trades and estimate portfolio state afterwards
        est_balances_after_trades = simulate_trades(
            trades_to_fiat,
            market_state,
        )
        est_values_after_trades = market_state.estimate_values(
            est_balances_after_trades,
            self.fiat,
        )

        fiat_after_trades = est_balances_after_trades[self.fiat]
        fiat_to_redistribute = fiat_after_trades - ideal_fiat_value_per_coin
        if fiat_to_redistribute <= 0:
            return trades_to_fiat

        # 3) Find coins in which we don't hold enough value
        possible_buys = set()
        for buy_coin in self._possible_investments(market_state):
            value = est_values_after_trades.get(buy_coin, 0)
            if ideal_fiat_value_per_coin > value:
                possible_buys.add(buy_coin)

        fiat_to_redistribute_per_coin = fiat_to_redistribute / len(possible_buys)

        # 4) Plan trades, fanning back out from fiat to others
        trades_from_fiat = []
        for buy_coin in sorted(possible_buys):
            value = est_values_after_trades.get(buy_coin, 0)
            delta = ideal_fiat_value_per_coin - value
            if delta > 0:
                available_fiat = min(fiat_to_redistribute_per_coin, delta)
                trades_from_fiat.append(
                    AbstractTrade(self.fiat, buy_coin, self.fiat, available_fiat),
                )

        return trades_to_fiat + trades_from_fiat